Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-25 05:26:47 +00:00
Code Activity

Made API more robust

Browse Source
This commit is contained in:
Paulus Schoutsen
2013-11-01 11:34:43 -07:00
parent 24b317f10d
commit 92f0cb20ff
4 changed files with 164 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
homeassistant/test.py
View File

@@ -15,54 +15,56 @@ import homeassistant as ha
import homeassistant.remote as remote
import homeassistant.httpinterface as hah
API_PASSWORD = "test1234"
HTTP_BASE_URL = "http://127.0.0.1:{}".format(hah.SERVER_PORT)
def _url(path=""):
""" Helper method to generate urls. """
return HTTP_BASE_URL + path
class HAHelper(object): # pylint: disable=too-few-public-methods
""" Helper class to keep track of current running HA instance. """
core = None
def ensure_homeassistant_started():
""" Ensures home assistant is started. """
if not HAHelper.core:
core = {'eventbus': ha.EventBus()}
core['statemachine'] = ha.StateMachine(core['eventbus'])
core['statemachine'].set_state('test','a_state')
hah.HTTPInterface(core['eventbus'], core['statemachine'],
API_PASSWORD)
core['eventbus'].fire(ha.EVENT_START)
# Give objects time to startup
time.sleep(1)
HAHelper.core = core
return HAHelper.core['eventbus'], HAHelper.core['statemachine']
# pylint: disable=too-many-public-methods
class TestHTTPInterface(unittest.TestCase):
""" Test the HTTP debug interface and API. """
HTTP_init = False
def _url(self, path=""):
""" Helper method to generate urls. """
return HTTP_BASE_URL + path
def setUp(self): # pylint: disable=invalid-name
""" Initialize the HTTP interface if not started yet. """
if not TestHTTPInterface.HTTP_init:
TestHTTPInterface.HTTP_init = True
hah.HTTPInterface(self.eventbus, self.statemachine, API_PASSWORD)
self.statemachine.set_state("test", "INIT_STATE")
self.sm_with_remote_eb.set_state("test", "INIT_STATE")
self.eventbus.fire(ha.EVENT_START)
# Give objects time to startup
time.sleep(1)
@classmethod
def setUpClass(cls): # pylint: disable=invalid-name
""" things to be run when tests are started. """
cls.eventbus = ha.EventBus()
cls.statemachine = ha.StateMachine(cls.eventbus)
cls.remote_sm = remote.StateMachine("127.0.0.1", API_PASSWORD)
cls.remote_eb = remote.EventBus("127.0.0.1", API_PASSWORD)
cls.sm_with_remote_eb = ha.StateMachine(cls.remote_eb)
cls.eventbus, cls.statemachine = ensure_homeassistant_started()
def test_debug_interface(self):
""" Test if we can login by comparing not logged in screen to
logged in screen. """
with_pw = requests.get(
self._url("/?api_password={}".format(API_PASSWORD)))
_url("/?api_password={}".format(API_PASSWORD)))
without_pw = requests.get(self._url())
without_pw = requests.get(_url())
self.assertNotEqual(without_pw.text, with_pw.text)
@@ -70,7 +72,7 @@ class TestHTTPInterface(unittest.TestCase):
def test_debug_state_change(self):
""" Test if the debug interface allows us to change a state. """
requests.post(
self._url(hah.URL_STATES_CATEGORY.format("test")),
_url(hah.URL_STATES_CATEGORY.format("test")),
data={"new_state":"debug_state_change",
"api_password":API_PASSWORD})
@@ -82,12 +84,12 @@ class TestHTTPInterface(unittest.TestCase):
""" Test if we get access denied if we omit or provide
a wrong api password. """
req = requests.post(
self._url(hah.URL_API_STATES_CATEGORY.format("test")))
_url(hah.URL_API_STATES_CATEGORY.format("test")))
self.assertEqual(req.status_code, 401)
req = requests.post(
self._url(hah.URL_API_STATES_CATEGORY.format("test")),
_url(hah.URL_API_STATES_CATEGORY.format("test")),
data={"api_password":"not the password"})
self.assertEqual(req.status_code, 401)
@@ -95,7 +97,7 @@ class TestHTTPInterface(unittest.TestCase):
def test_api_list_state_categories(self):
""" Test if the debug interface allows us to list state categories. """
req = requests.get(self._url(hah.URL_API_STATES),
req = requests.get(_url(hah.URL_API_STATES),
data={"api_password":API_PASSWORD})
data = req.json()
@@ -107,7 +109,7 @@ class TestHTTPInterface(unittest.TestCase):
def test_api_get_state(self):
""" Test if the debug interface allows us to get a state. """
req = requests.get(
self._url(hah.URL_API_STATES_CATEGORY.format("test")),
_url(hah.URL_API_STATES_CATEGORY.format("test")),
data={"api_password":API_PASSWORD})
data = req.json()
@@ -119,50 +121,26 @@ class TestHTTPInterface(unittest.TestCase):
self.assertEqual(data['last_changed'], state['last_changed'])
self.assertEqual(data['attributes'], state['attributes'])
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_CATEGORY.format("does_not_exist")),
data={"api_password":API_PASSWORD})
self.assertEqual(req.status_code, 422)
def test_api_state_change(self):
""" Test if we can change the state of a category that exists. """
self.statemachine.set_state("test", "not_to_be_set_state")
requests.post(self._url(hah.URL_API_STATES_CATEGORY.format("test")),
requests.post(_url(hah.URL_API_STATES_CATEGORY.format("test")),
data={"new_state":"debug_state_change2",
"api_password":API_PASSWORD})
self.assertEqual(self.statemachine.get_state("test")['state'],
"debug_state_change2")
# pylint: disable=invalid-name
def test_remote_sm_list_state_categories(self):
""" Test if the debug interface allows us to list state categories. """
self.assertEqual(self.statemachine.categories,
self.remote_sm.categories)
def test_remote_sm_get_state(self):
""" Test if the debug interface allows us to list state categories. """
remote_state = self.remote_sm.get_state("test")
state = self.statemachine.get_state("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_state_change(self):
""" Test if we can change the state of a category that exists. """
self.remote_sm.set_state("test", "set_remotely", {"test": 1})
state = self.statemachine.get_state("test")
self.assertEqual(state['state'], "set_remotely")
self.assertEqual(state['attributes']['test'], 1)
# pylint: disable=invalid-name
def test_api_state_change_of_non_existing_category(self):
""" Test if the API allows us to change a state of
@@ -171,7 +149,7 @@ class TestHTTPInterface(unittest.TestCase):
new_state = "debug_state_change"
req = requests.post(
self._url(hah.URL_API_STATES_CATEGORY.format(
_url(hah.URL_API_STATES_CATEGORY.format(
"test_category_that_does_not_exist")),
data={"new_state": new_state,
"api_password": API_PASSWORD})
@@ -194,7 +172,7 @@ class TestHTTPInterface(unittest.TestCase):
self.eventbus.listen_once("test_event_no_data", listener)
requests.post(
self._url(hah.URL_EVENTS_EVENT.format("test_event_no_data")),
_url(hah.URL_EVENTS_EVENT.format("test_event_no_data")),
data={"api_password":API_PASSWORD})
# Allow the event to take place
@@ -216,7 +194,7 @@ class TestHTTPInterface(unittest.TestCase):
self.eventbus.listen_once("test_event_with_data", listener)
requests.post(
self._url(hah.URL_EVENTS_EVENT.format("test_event_with_data")),
_url(hah.URL_EVENTS_EVENT.format("test_event_with_data")),
data={"event_data":'{"test": 1}',
"api_password":API_PASSWORD})
@@ -238,7 +216,7 @@ class TestHTTPInterface(unittest.TestCase):
self.eventbus.listen_once("test_event_with_bad_data", listener)
req = requests.post(
self._url(hah.URL_API_EVENTS_EVENT.format("test_event")),
_url(hah.URL_API_EVENTS_EVENT.format("test_event")),
data={"event_data":'not json',
"api_password":API_PASSWORD})
@@ -246,10 +224,57 @@ class TestHTTPInterface(unittest.TestCase):
# It shouldn't but if it fires, allow the event to take place
time.sleep(1)
self.assertEqual(req.status_code, 400)
self.assertEqual(req.status_code, 422)
self.assertEqual(len(test_value), 0)
class TestRemote(unittest.TestCase):
""" Test the homeassistant.remote module. """
@classmethod
def setUpClass(cls): # pylint: disable=invalid-name
""" things to be run when tests are started. """
cls.eventbus, cls.statemachine = 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.sm_with_remote_eb = ha.StateMachine(cls.remote_eb)
cls.sm_with_remote_eb.set_state("test", "a_state")
# pylint: disable=invalid-name
def test_remote_sm_list_state_categories(self):
""" Test if the debug interface allows us to list state categories. """
self.assertEqual(self.statemachine.categories,
self.remote_sm.categories)
def test_remote_sm_get_state(self):
""" Test if the debug interface allows us to list state categories. """
remote_state = self.remote_sm.get_state("test")
state = self.statemachine.get_state("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 if the debug interface allows us to list state categories. """
self.assertEqual(self.remote_sm.get_state("test_does_not_exist"), None)
def test_remote_sm_state_change(self):
""" Test if we can change the state of a category that exists. """
self.remote_sm.set_state("test", "set_remotely", {"test": 1})
state = self.statemachine.get_state("test")
self.assertEqual(state['state'], "set_remotely")
self.assertEqual(state['attributes']['test'], 1)
# pylint: disable=invalid-name
def test_remote_eb_fire_event_with_no_data(self):
""" Test if the remote eventbus allows us to fire an event. """
test_value = []
@@ -303,4 +328,3 @@ class TestHTTPInterface(unittest.TestCase):
time.sleep(1)
self.assertEqual(len(test_value), 1)