Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-23 04:19:34 +00:00
Code Activity

Remote instances are now 100% operational

Browse Source
This commit is contained in:
Paulus Schoutsen
2014-04-29 00:30:31 -07:00
parent 8e65afa994
commit 50b492c64a
12 changed files with 770 additions and 529 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
homeassistant/test.py
View File

@@ -13,11 +13,11 @@ import requests
import homeassistant as ha
import homeassistant.remote as remote
import homeassistant.components.httpinterface as hah
import homeassistant.components.http as http
API_PASSWORD = "test1234"
HTTP_BASE_URL = "http://127.0.0.1:{}".format(hah.SERVER_PORT)
HTTP_BASE_URL = "http://127.0.0.1:{}".format(remote.SERVER_PORT)
def _url(path=""):
@@ -28,6 +28,7 @@ def _url(path=""):
class HAHelper(object): # pylint: disable=too-few-public-methods
""" Helper class to keep track of current running HA instance. """
hass = None
slave = None
def ensure_homeassistant_started():
@@ -39,9 +40,9 @@ def ensure_homeassistant_started():
hass.bus.listen('test_event', len)
hass.states.set('test', 'a_state')
hah.HTTPInterface(hass, API_PASSWORD)
http.setup(hass, API_PASSWORD)
hass.bus.fire(ha.EVENT_HOMEASSISTANT_START)
hass.start()
# Give objects time to startup
time.sleep(1)
@@ -51,6 +52,26 @@ def ensure_homeassistant_started():
return HAHelper.hass
def ensure_slave_started():
""" Ensure a home assistant slave is started. """
if not HAHelper.slave:
local_api = remote.API("127.0.0.1", API_PASSWORD, 8124)
remote_api = remote.API("127.0.0.1", API_PASSWORD)
slave = remote.HomeAssistant(local_api, remote_api)
http.setup(slave, API_PASSWORD, 8124)
slave.start()
# Give objects time to startup
time.sleep(1)
HAHelper.slave = slave
return HAHelper.slave
# pylint: disable=too-many-public-methods
class TestHTTPInterface(unittest.TestCase):
""" Test the HTTP debug interface and API. """
@@ -75,12 +96,12 @@ class TestHTTPInterface(unittest.TestCase):
""" Test if we get access denied if we omit or provide
a wrong api password. """
req = requests.get(
_url(hah.URL_API_STATES_ENTITY.format("test")))
_url(remote.URL_API_STATES_ENTITY.format("test")))
self.assertEqual(req.status_code, 401)
req = requests.get(
_url(hah.URL_API_STATES_ENTITY.format("test")),
_url(remote.URL_API_STATES_ENTITY.format("test")),
params={"api_password": "not the password"})
self.assertEqual(req.status_code, 401)
@@ -89,7 +110,7 @@ class TestHTTPInterface(unittest.TestCase):
""" Test if we can change a state from the debug interface. """
self.hass.states.set("test.test", "not_to_be_set")
requests.post(_url(hah.URL_CHANGE_STATE),
requests.post(_url(http.URL_CHANGE_STATE),
data={"entity_id": "test.test",
"new_state": "debug_state_change2",
"api_password": API_PASSWORD})
@@ -110,7 +131,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.listen_once_event("test_event_with_data", listener)
requests.post(
_url(hah.URL_FIRE_EVENT),
_url(http.URL_FIRE_EVENT),
data={"event_type": "test_event_with_data",
"event_data": '{"test": 1}',
"api_password": API_PASSWORD})
@@ -122,18 +143,20 @@ class TestHTTPInterface(unittest.TestCase):
def test_api_list_state_entities(self):
""" Test if the debug interface allows us to list state entities. """
req = requests.get(_url(hah.URL_API_STATES),
req = requests.get(_url(remote.URL_API_STATES),
data={"api_password": API_PASSWORD})
data = req.json()
remote_data = req.json()
self.assertEqual(list(self.hass.states.entity_ids),
data['entity_ids'])
local_data = {entity_id: state.as_dict() for entity_id, state
in self.hass.states.all().items()}
self.assertEqual(local_data, remote_data)
def test_api_get(self):
""" Test if the debug interface allows us to get a state. """
req = requests.get(
_url(hah.URL_API_STATES_ENTITY.format("test")),
_url(remote.URL_API_STATES_ENTITY.format("test")),
data={"api_password": API_PASSWORD})
data = ha.State.from_dict(req.json())
@@ -147,7 +170,7 @@ class TestHTTPInterface(unittest.TestCase):
def test_api_get_non_existing_state(self):
""" Test if the debug interface allows us to get a state. """
req = requests.get(
_url(hah.URL_API_STATES_ENTITY.format("does_not_exist")),
_url(remote.URL_API_STATES_ENTITY.format("does_not_exist")),
params={"api_password": API_PASSWORD})
self.assertEqual(req.status_code, 422)
@@ -157,7 +180,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.states.set("test.test", "not_to_be_set")
requests.post(_url(hah.URL_API_STATES_ENTITY.format("test.test")),
requests.post(_url(remote.URL_API_STATES_ENTITY.format("test.test")),
data={"new_state": "debug_state_change2",
"api_password": API_PASSWORD})
@@ -172,7 +195,7 @@ class TestHTTPInterface(unittest.TestCase):
new_state = "debug_state_change"
req = requests.post(
_url(hah.URL_API_STATES_ENTITY.format(
_url(remote.URL_API_STATES_ENTITY.format(
"test_entity_that_does_not_exist")),
data={"new_state": new_state,
"api_password": API_PASSWORD})
@@ -195,7 +218,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.listen_once_event("test.event_no_data", listener)
requests.post(
_url(hah.URL_API_EVENTS_EVENT.format("test.event_no_data")),
_url(remote.URL_API_EVENTS_EVENT.format("test.event_no_data")),
data={"api_password": API_PASSWORD})
# Allow the event to take place
@@ -217,7 +240,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.listen_once_event("test_event_with_data", listener)
requests.post(
_url(hah.URL_API_EVENTS_EVENT.format("test_event_with_data")),
_url(remote.URL_API_EVENTS_EVENT.format("test_event_with_data")),
data={"event_data": '{"test": 1}',
"api_password": API_PASSWORD})
@@ -238,7 +261,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.listen_once_event("test_event_with_bad_data", listener)
req = requests.post(
_url(hah.URL_API_EVENTS_EVENT.format("test_event")),
_url(remote.URL_API_EVENTS_EVENT.format("test_event")),
data={"event_data": 'not json',
"api_password": API_PASSWORD})
@@ -250,7 +273,7 @@ class TestHTTPInterface(unittest.TestCase):
def test_api_get_event_listeners(self):
""" Test if we can get the list of events being listened for. """
req = requests.get(_url(hah.URL_API_EVENTS),
req = requests.get(_url(remote.URL_API_EVENTS),
params={"api_password": API_PASSWORD})
data = req.json()
@@ -259,7 +282,7 @@ class TestHTTPInterface(unittest.TestCase):
def test_api_get_services(self):
""" Test if we can get a dict describing current services. """
req = requests.get(_url(hah.URL_API_SERVICES),
req = requests.get(_url(remote.URL_API_SERVICES),
params={"api_password": API_PASSWORD})
data = req.json()
@@ -277,7 +300,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.services.register("test_domain", "test_service", listener)
requests.post(
_url(hah.URL_API_SERVICES_SERVICE.format(
_url(remote.URL_API_SERVICES_SERVICE.format(
"test_domain", "test_service")),
data={"api_password": API_PASSWORD})
@@ -299,7 +322,7 @@ class TestHTTPInterface(unittest.TestCase):
self.hass.services.register("test_domain", "test_service", listener)
requests.post(
_url(hah.URL_API_SERVICES_SERVICE.format(
_url(remote.URL_API_SERVICES_SERVICE.format(
"test_domain", "test_service")),
data={"service_data": '{"test": 1}',
"api_password": API_PASSWORD})
@@ -310,7 +333,7 @@ class TestHTTPInterface(unittest.TestCase):
self.assertEqual(len(test_value), 1)
class TestRemote(unittest.TestCase):
class TestRemoteMethods(unittest.TestCase):
""" Test the homeassistant.remote module. """
@classmethod
@@ -318,134 +341,115 @@ class TestRemote(unittest.TestCase):
""" things to be run when tests are started. """
cls.hass = ensure_homeassistant_started()
cls.remote_sm = remote.StateMachine("127.0.0.1", API_PASSWORD)
cls.remote_eb = remote.EventBus("127.0.0.1", API_PASSWORD)
cls.remote_sr = remote.ServiceRegistry("127.0.0.1", API_PASSWORD)
cls.sm_with_remote_eb = ha.StateMachine(cls.remote_eb)
cls.sm_with_remote_eb.set("test", "a_state")
cls.api = remote.API("127.0.0.1", API_PASSWORD)
# pylint: disable=invalid-name
def test_remote_sm_list_state_entities(self):
""" Test if the debug interface allows us to list state entity ids. """
def test_get_event_listeners(self):
""" Test Python API get_event_listeners. """
self.assertEqual(list(self.hass.states.entity_ids),
self.remote_sm.entity_ids)
self.assertEqual(
remote.get_event_listeners(self.api), self.hass.bus.listeners)
def test_remote_sm_get(self):
""" Test if debug interface allows us to get state of an entity. """
remote_state = self.remote_sm.get("test")
state = self.hass.states.get("test")
self.assertEqual(remote_state.state, state.state)
self.assertEqual(remote_state.last_changed, state.last_changed)
self.assertEqual(remote_state.attributes, state.attributes)
def test_remote_sm_get_non_existing_state(self):
""" Test remote state machine to get state of non existing entity. """
self.assertEqual(self.remote_sm.get("test_does_not_exist"), None)
def test_remote_sm_state_change(self):
""" Test if we can change the state of an existing entity. """
self.remote_sm.set("test", "set_remotely", {"test": 1})
state = self.hass.states.get("test")
self.assertEqual(state.state, "set_remotely")
self.assertEqual(state.attributes['test'], 1)
def test_remote_eb_listening_for_same(self):
""" Test if remote EB correctly reports listener overview. """
self.assertEqual(self.hass.bus.listeners,
self.remote_eb.listeners)
# pylint: disable=invalid-name
def test_remote_eb_fire_event_with_no_data(self):
""" Test if the remote bus allows us to fire an event. """
def test_fire_event(self):
""" Test Python API fire_event. """
test_value = []
def listener(event): # pylint: disable=unused-argument
""" Helper method that will verify our event got called. """
test_value.append(1)
self.hass.listen_once_event("test_event_no_data", listener)
self.hass.listen_once_event("test.event_no_data", listener)
self.remote_eb.fire("test_event_no_data")
remote.fire_event(self.api, "test.event_no_data")
# Allow the event to take place
time.sleep(1)
self.assertEqual(len(test_value), 1)
# pylint: disable=invalid-name
def test_remote_eb_fire_event_with_data(self):
""" Test if the remote bus allows us to fire an event. """
test_value = []
def test_get_state(self):
""" Test Python API get_state. """
def listener(event): # pylint: disable=unused-argument
""" Helper method that will verify our event got called. """
if event.data["test"] == 1:
test_value.append(1)
self.assertEqual(
remote.get_state(self.api, 'test'), self.hass.states.get('test'))
self.hass.listen_once_event("test_event_with_data", listener)
def test_get_states(self):
""" Test Python API get_state_entity_ids. """
self.remote_eb.fire("test_event_with_data", {"test": 1})
self.assertEqual(
remote.get_states(self.api), self.hass.states.all())
# Allow the event to take place
time.sleep(1)
def test_set_state(self):
""" Test Python API set_state. """
remote.set_state(self.api, 'test', 'set_test')
self.assertEqual(len(test_value), 1)
self.assertEqual(self.hass.states.get('test').state, 'set_test')
# pylint: disable=invalid-name
def test_remote_sr_call_service_with_no_data(self):
""" Test if the remote bus allows us to fire a service. """
def test_is_state(self):
""" Test Python API is_state. """
self.assertEqual(
remote.is_state(self.api, 'test',
self.hass.states.get('test').state),
True)
def test_get_services(self):
""" Test Python API get_services. """
self.assertEqual(
remote.get_services(self.api), self.hass.services.services)
def test_call_service(self):
""" Test Python API call_service. """
test_value = []
def listener(service_call): # pylint: disable=unused-argument
""" Helper method that will verify our service got called. """
""" Helper method that will verify that our service got called. """
test_value.append(1)
self.hass.services.register("test_domain", "test_service", listener)
self.remote_sr.call_service("test_domain", "test_service")
# Allow the service call to take place
time.sleep(1)
self.assertEqual(len(test_value), 1)
# pylint: disable=invalid-name
def test_remote_sr_call_service_with_data(self):
""" Test if the remote bus allows us to fire an event. """
test_value = []
def listener(service_call): # pylint: disable=unused-argument
""" Helper method that will verify our service got called. """
if service_call.data["test"] == 1:
test_value.append(1)
self.hass.services.register("test_domain", "test_service", listener)
self.remote_sr.call_service("test_domain", "test_service", {"test": 1})
remote.call_service(self.api, "test_domain", "test_service")
# Allow the event to take place
time.sleep(1)
self.assertEqual(len(test_value), 1)
def test_local_sm_with_remote_eb(self):
""" Test if we get the event if we change a state on a
StateMachine connected to a remote bus. """
class TestRemoteClasses(unittest.TestCase):
""" Test the homeassistant.remote module. """
@classmethod
def setUpClass(cls): # pylint: disable=invalid-name
""" things to be run when tests are started. """
cls.hass = ensure_homeassistant_started()
cls.slave = ensure_slave_started()
def test_statemachine_init(self):
""" Tests if remote.StateMachine copies all states on init. """
self.assertEqual(self.hass.states.all(), self.slave.states.all())
def test_statemachine_set(self):
""" Tests if setting the state on a slave is recorded. """
self.slave.states.set("test", "remote.statemachine test")
# Allow interaction between 2 instances
time.sleep(1)
self.assertEqual(self.slave.states.get("test").state,
"remote.statemachine test")
def test_eventbus_fire(self):
""" Test if events fired from the eventbus get fired. """
test_value = []
def listener(event): # pylint: disable=unused-argument
""" Helper method that will verify our event got called. """
test_value.append(1)
self.hass.listen_once_event(ha.EVENT_STATE_CHANGED, listener)
self.slave.listen_once_event("test.event_no_data", listener)
self.sm_with_remote_eb.set("test", "local sm with remote eb")
self.slave.bus.fire("test.event_no_data")
# Allow the event to take place
time.sleep(1)