diff --git a/homeassistant/test.py b/homeassistant/test.py index e60da7f67db..d641c20e2fd 100644 --- a/homeassistant/test.py +++ b/homeassistant/test.py @@ -5,7 +5,7 @@ homeassistant.test Provides tests to verify that Home Assistant modules do what they should do. """ - +import os import unittest import time import json @@ -80,6 +80,137 @@ def ensure_slave_started(): return HAHelper.slave +# pylint: disable=too-many-public-methods +class TestHomeAssistant(unittest.TestCase): + """ + Tests the Home Assistant core classes. + Currently only includes tests to test cases that do not + get tested in the API integration tests. + """ + + def setUp(self): # pylint: disable=invalid-name + """ things to be run when tests are started. """ + self.hass = ha.HomeAssistant() + self.hass.states.set("light.Bowl", "on") + self.hass.states.set("switch.AC", "off") + + def test_get_config_path(self): + """ Test get_config_path method. """ + self.assertEqual(os.getcwd(), self.hass.config_dir) + + self.assertEqual(os.path.join(os.getcwd(), "test.conf"), + self.hass.get_config_path("test.conf")) + + def test_block_till_stoped(self): + """ Test if we can block till stop service is called. """ + pass + + def test_get_entity_ids(self): + """ Test get_entity_ids method. """ + ent_ids = self.hass.get_entity_ids() + self.assertEqual(2, len(ent_ids)) + self.assertTrue('light.Bowl' in ent_ids) + self.assertTrue('switch.AC' in ent_ids) + + ent_ids = self.hass.get_entity_ids('light') + self.assertEqual(1, len(ent_ids)) + self.assertTrue('light.Bowl' in ent_ids) + + def test_track_state_change(self): + """ Test track_state_change. """ + # with with from_state, to_state and without + pass + + def test_track_point_in_time(self): + """ Test track point in time. """ + pass + + def test_track_time_change(self): + """ Test tracking time change. """ + # with paramters + # without parameters + pass + + +# pylint: disable=too-many-public-methods +class TestEventBus(unittest.TestCase): + """ Test EventBus methods. """ + + def setUp(self): # pylint: disable=invalid-name + """ things to be run when tests are started. """ + self.bus = ha.EventBus() + self.bus.listen('test_event', lambda x: len) + + def test_add_remove_listener(self): + """ Test remove_listener method. """ + old_count = len(self.bus.listeners) + + listener = lambda x: len + + self.bus.listen('test', listener) + + self.assertEqual(old_count + 1, len(self.bus.listeners)) + + self.bus.remove_listener('test', listener) + + self.assertEqual(old_count, len(self.bus.listeners)) + + +# pylint: disable=too-many-public-methods +class TestState(unittest.TestCase): + """ Test EventBus methods. """ + + def test_init(self): + """ Test state.init """ + self.assertRaises( + ha.InvalidEntityFormatError, ha.State, + 'invalid_entity_format', 'test_state') + + +# pylint: disable=too-many-public-methods +class TestStateMachine(unittest.TestCase): + """ Test EventBus methods. """ + + def setUp(self): # pylint: disable=invalid-name + """ things to be run when tests are started. """ + self.bus = ha.EventBus() + self.states = ha.StateMachine(self.bus) + self.states.set("light.Bowl", "on") + self.states.set("switch.AC", "off") + + def test_is_state(self): + """ Test is_state method. """ + self.assertTrue(self.states.is_state('light.Bowl', 'on')) + self.assertFalse(self.states.is_state('light.Bowl', 'off')) + self.assertFalse(self.states.is_state('light.Non_existing', 'on')) + + def test_remove(self): + """ Test remove method. """ + self.assertTrue('light.Bowl' in self.states.entity_ids) + self.assertTrue(self.states.remove('light.Bowl')) + self.assertFalse('light.Bowl' in self.states.entity_ids) + + # If it does not exist, we should get False + self.assertFalse(self.states.remove('light.Bowl')) + + +# pylint: disable=too-many-public-methods +class TestServiceRegistry(unittest.TestCase): + """ Test EventBus methods. """ + + def setUp(self): # pylint: disable=invalid-name + """ things to be run when tests are started. """ + pool = ha.create_worker_pool() + self.bus = ha.EventBus(pool) + self.services = ha.ServiceRegistry(self.bus, pool) + self.services.register("test_domain", "test_service", lambda x: len) + + def test_has_service(self): + """ Test has_service method. """ + self.assertTrue( + self.services.has_service("test_domain", "test_service")) + + # pylint: disable=too-many-public-methods class TestHTTP(unittest.TestCase): """ Test the HTTP debug interface and API. """ @@ -95,13 +226,13 @@ class TestHTTP(unittest.TestCase): req = requests.get( _url(remote.URL_API_STATES_ENTITY.format("test"))) - self.assertEqual(req.status_code, 401) + self.assertEqual(401, req.status_code) req = requests.get( _url(remote.URL_API_STATES_ENTITY.format("test")), headers={remote.AUTH_HEADER: 'wrongpassword'}) - self.assertEqual(req.status_code, 401) + self.assertEqual(401, req.status_code) def test_api_list_state_entities(self): """ Test if the debug interface allows us to list state entities. """ @@ -122,9 +253,9 @@ class TestHTTP(unittest.TestCase): state = self.hass.states.get("test.test") - self.assertEqual(data.state, state.state) - self.assertEqual(data.last_changed, state.last_changed) - self.assertEqual(data.attributes, state.attributes) + self.assertEqual(state.state, data.state) + self.assertEqual(state.last_changed, data.last_changed) + self.assertEqual(state.attributes, data.attributes) def test_api_get_non_existing_state(self): """ Test if the debug interface allows us to get a state. """ @@ -132,7 +263,7 @@ class TestHTTP(unittest.TestCase): _url(remote.URL_API_STATES_ENTITY.format("does_not_exist")), headers=HA_HEADERS) - self.assertEqual(req.status_code, 404) + self.assertEqual(404, req.status_code) def test_api_state_change(self): """ Test if we can change the state of an entity that exists. """ @@ -143,8 +274,8 @@ class TestHTTP(unittest.TestCase): data=json.dumps({"state": "debug_state_change2", "api_password": API_PASSWORD})) - self.assertEqual(self.hass.states.get("test.test").state, - "debug_state_change2") + self.assertEqual("debug_state_change2", + self.hass.states.get("test.test").state) # pylint: disable=invalid-name def test_api_state_change_of_non_existing_entity(self): @@ -162,7 +293,7 @@ class TestHTTP(unittest.TestCase): cur_state = (self.hass.states. get("test_entity.that_does_not_exist").state) - self.assertEqual(req.status_code, 201) + self.assertEqual(201, req.status_code) self.assertEqual(cur_state, new_state) # pylint: disable=invalid-name @@ -183,7 +314,7 @@ class TestHTTP(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value)) # pylint: disable=invalid-name def test_api_fire_event_with_data(self): @@ -206,7 +337,7 @@ class TestHTTP(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value)) # pylint: disable=invalid-name def test_api_fire_event_with_invalid_json(self): @@ -227,8 +358,8 @@ class TestHTTP(unittest.TestCase): # It shouldn't but if it fires, allow the event to take place time.sleep(1) - self.assertEqual(req.status_code, 422) - self.assertEqual(len(test_value), 0) + self.assertEqual(422, req.status_code) + self.assertEqual(0, len(test_value)) def test_api_get_event_listeners(self): """ Test if we can get the list of events being listened for. """ @@ -241,7 +372,7 @@ class TestHTTP(unittest.TestCase): self.assertEqual(event["listener_count"], local.pop(event["event"])) - self.assertEqual(len(local), 0) + self.assertEqual(0, len(local)) def test_api_get_services(self): """ Test if we can get a dict describing current services. """ @@ -253,7 +384,7 @@ class TestHTTP(unittest.TestCase): for serv_domain in req.json(): local = local_services.pop(serv_domain["domain"]) - self.assertEqual(serv_domain["services"], local) + self.assertEqual(local, serv_domain["services"]) def test_api_call_service_no_data(self): """ Test if the API allows us to call a service. """ @@ -273,7 +404,7 @@ class TestHTTP(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value)) def test_api_call_service_with_data(self): """ Test if the API allows us to call a service. """ @@ -296,7 +427,7 @@ class TestHTTP(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value)) class TestRemoteMethods(unittest.TestCase): @@ -311,13 +442,14 @@ class TestRemoteMethods(unittest.TestCase): def test_get_event_listeners(self): """ Test Python API get_event_listeners. """ - local = self.hass.bus.listeners + local_data = self.hass.bus.listeners + remote_data = remote.get_event_listeners(self.api) - for event in remote.get_event_listeners(self.api): - self.assertEqual(event["listener_count"], - local.pop(event["event"])) + for event in remote_data: + self.assertEqual(local_data.pop(event["event"]), + event["listener_count"]) - self.assertEqual(len(local), 0) + self.assertEqual(len(local_data), 0) def test_fire_event(self): """ Test Python API fire_event. """ @@ -334,14 +466,14 @@ class TestRemoteMethods(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value)) def test_get_state(self): """ Test Python API get_state. """ self.assertEqual( - remote.get_state(self.api, 'test.test'), - self.hass.states.get('test.test')) + self.hass.states.get('test.test'), + remote.get_state(self.api, 'test.test')) def test_get_states(self): """ Test Python API get_state_entity_ids. """ @@ -353,7 +485,7 @@ class TestRemoteMethods(unittest.TestCase): """ Test Python API set_state. """ self.assertTrue(remote.set_state(self.api, 'test.test', 'set_test')) - self.assertEqual(self.hass.states.get('test.test').state, 'set_test') + self.assertEqual('set_test', self.hass.states.get('test.test').state) def test_is_state(self): """ Test Python API is_state. """ @@ -370,7 +502,7 @@ class TestRemoteMethods(unittest.TestCase): for serv_domain in remote.get_services(self.api): local = local_services.pop(serv_domain["domain"]) - self.assertEqual(serv_domain["services"], local) + self.assertEqual(local, serv_domain["services"]) def test_call_service(self): """ Test Python API call_service. """ @@ -387,7 +519,7 @@ class TestRemoteMethods(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value)) class TestRemoteClasses(unittest.TestCase): @@ -401,9 +533,12 @@ class TestRemoteClasses(unittest.TestCase): def test_statemachine_init(self): """ Tests if remote.StateMachine copies all states on init. """ + self.assertEqual(len(self.hass.states.all()), + len(self.slave.states.all())) + for state in self.hass.states.all(): self.assertEqual( - self.slave.states.get(state.entity_id), state) + state, self.slave.states.get(state.entity_id)) def test_statemachine_set(self): """ Tests if setting the state on a slave is recorded. """ @@ -412,8 +547,8 @@ class TestRemoteClasses(unittest.TestCase): # Allow interaction between 2 instances time.sleep(1) - self.assertEqual(self.slave.states.get("remote.test").state, - "remote.statemachine test") + self.assertEqual("remote.statemachine test", + self.slave.states.get("remote.test").state) def test_eventbus_fire(self): """ Test if events fired from the eventbus get fired. """ @@ -430,4 +565,4 @@ class TestRemoteClasses(unittest.TestCase): # Allow the event to take place time.sleep(1) - self.assertEqual(len(test_value), 1) + self.assertEqual(1, len(test_value))