<commit_before># -*- coding: utf-8 -*-

from __future__ import absolute_import, division, print_function, unicode_literals

import time
from unittest import TestCase
from . import SB_TEST_HOST, SB_TEST_PORT

from pysensorbee.api import SensorBeeAPI, ResultSet, SensorBeeAPIError, WebSocketClient


class SensorBeeAPITest(TestCase):
    TOPOLOGY = 'test'
    TOPOLOGY2 = 'test2'

    def setUp(self):
        self.api = SensorBeeAPI(SB_TEST_HOST, SB_TEST_PORT)
        self.api.delete_topology(self.TOPOLOGY)
        self.api.create_topology(self.TOPOLOGY)

    def tearDown(self):
        self.api.delete_topology(self.TOPOLOGY)
        self.api.delete_topology(self.TOPOLOGY2)

    def test_error(self):
        self.assertRaises(SensorBeeAPIError, self.api.sources, 'no_such_topology')
        self.assertRaises(SensorBeeAPIError, self.api.query, self.TOPOLOGY, 'INVALID BQL')

    def test_init(self):
        self.assertEqual(SB_TEST_HOST, self.api.host)
        self.assertEqual(SB_TEST_PORT, self.api.port)

    def test_runtime_status(self):
        status = self.api.runtime_status()
        self.assertTrue(isinstance(status, dict))
        self.assertTrue('pid' in status)

    def test_topologies(self):
        api = self.api
        self.assertEqual(1, len(api.topologies()['topologies']))

        self.assertEqual(self.TOPOLOGY2,
            api.create_topology(self.TOPOLOGY2)['topology']['name'])
        try:
            self.assertEqual(2, len(api.topologies()['topologies']))
        finally:
            self.assertEqual({}, api.delete_topology(self.TOPOLOGY2))
            self.assertEqual(1, len(api.topologies()['topologies']))

    def test_sources(self):
        api = self.api<commit_msg>add topology API test<commit_after># -*- coding: utf-8 -*-

from __future__ import absolute_import, division, print_function, unicode_literals

import time
from unittest import TestCase
from . import SB_TEST_HOST, SB_TEST_PORT

from pysensorbee.api import SensorBeeAPI, ResultSet, SensorBeeAPIError, WebSocketClient


class SensorBeeAPITest(TestCase):
    TOPOLOGY = 'test'
    TOPOLOGY2 = 'test2'

    def setUp(self):
        self.api = SensorBeeAPI(SB_TEST_HOST, SB_TEST_PORT)
        self.api.delete_topology(self.TOPOLOGY)
        self.api.create_topology(self.TOPOLOGY)

    def tearDown(self):
        self.api.delete_topology(self.TOPOLOGY)
        self.api.delete_topology(self.TOPOLOGY2)

    def test_error(self):
        self.assertRaises(SensorBeeAPIError, self.api.sources, 'no_such_topology')
        self.assertRaises(SensorBeeAPIError, self.api.query, self.TOPOLOGY, 'INVALID BQL')

    def test_init(self):
        self.assertEqual(SB_TEST_HOST, self.api.host)
        self.assertEqual(SB_TEST_PORT, self.api.port)

    def test_runtime_status(self):
        status = self.api.runtime_status()
        self.assertTrue(isinstance(status, dict))
        self.assertTrue('pid' in status)

    def test_topologies(self):
        api = self.api
        self.assertEqual(1, len(api.topologies()['topologies']))

        self.assertEqual(self.TOPOLOGY2,
            api.create_topology(self.TOPOLOGY2)['topology']['name'])
        try:
            self.assertEqual(self.TOPOLOGY2, api.topology(self.TOPOLOGY2)['topology']['name'])
            self.assertEqual(2, len(api.topologies()['topologies']))
        finally:
            self.assertEqual({}, api.delete_topology(self.TOPOLOGY2))
            self.assertEqual(1, len(api.topologies()['topologies']))

    def test_sources(self):
        api = self.api
{'fraction_non_alphanumeric': 0.11576563826776111, 'fraction_numerical': 0.006227002547410133, 'mean_word_length': 3.4064837905236907, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2857938', 'n_tokens_mistral': 1178, 'n_tokens_neox': 1140, 'n_words': 182}