jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reorganized testing

Browse Source
This commit is contained in:
Paulus Schoutsen 2014-11-23 12:57:29 -08:00
parent ad16c32504
commit bc4b81d525
15 changed files with 259 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.travis.yml
View File

@ -7,6 +7,6 @@ install:
script:
- flake8 homeassistant --exclude bower_components,external
- pylint homeassistant
- coverage run --source=homeassistant -m homeassistant -t test
- coverage run --source=homeassistant -m unittest discover test
after_success:
- coveralls

29
homeassistant/__init__.py
View File

@ -24,6 +24,7 @@ DOMAIN = "homeassistant"
SERVICE_HOMEASSISTANT_STOP = "stop"
EVENT_HOMEASSISTANT_START = "homeassistant_start"
EVENT_HOMEASSISTANT_STOP = "homeassistant_stop"
EVENT_STATE_CHANGED = "state_changed"
EVENT_TIME_CHANGED = "time_changed"
EVENT_CALL_SERVICE = "call_service"
@ -62,10 +63,6 @@ class HomeAssistant(object):
self.services = ServiceRegistry(self.bus, pool)
self.states = StateMachine(self.bus)
# Components in a thread we might want to stop later
self.timer = None
self.http = None
self.config_dir = os.getcwd()
def get_config_path(self, path):
@ -74,7 +71,7 @@ class HomeAssistant(object):
def start(self):
""" Start home assistant. """
self.timer = Timer(self)
Timer(self)
self.bus.fire(EVENT_HOMEASSISTANT_START)
@ -234,14 +231,13 @@ class HomeAssistant(object):
""" Stops Home Assistant and shuts down all threads. """
_LOGGER.info("Stopping")
self.bus.fire(EVENT_HOMEASSISTANT_STOP)
# Wait till all responses to homeassistant_stop are done
self._pool.block_till_done()
self._pool.stop()
if self.http is not None:
self.http.shutdown()
if self.timer is not None:
self.timer.shutdown()
def _process_match_param(parameter):
""" Wraps parameter in a list if it is not one and returns it. """
@ -263,7 +259,7 @@ def _matcher(subject, pattern):
class JobPriority(util.OrderedEnum):
""" Provides priorities for bus events. """
# pylint: disable=no-init
# pylint: disable=no-init,too-few-public-methods
EVENT_SERVICE = 1
EVENT_STATE = 2
@ -312,7 +308,7 @@ def create_worker_pool(thread_count=POOL_NUM_THREAD):
class EventOrigin(enum.Enum):
""" Distinguish between origin of event. """
# pylint: disable=no-init
# pylint: disable=no-init,too-few-public-methods
local = "LOCAL"
remote = "REMOTE"
@ -626,6 +622,9 @@ class Timer(threading.Thread):
hass.listen_once_event(EVENT_HOMEASSISTANT_START,
lambda event: self.start())
hass.listen_once_event(EVENT_HOMEASSISTANT_STOP,
lambda event: self._stop.set())
def run(self):
""" Start the timer. """
@ -661,10 +660,6 @@ class Timer(threading.Thread):
self._bus.fire(EVENT_TIME_CHANGED, {ATTR_NOW: now})
def shutdown(self):
_LOGGER.info("Timer:Stopping")
self._stop.set()
class HomeAssistantError(Exception):
""" General Home Assistant exception occured. """

82
homeassistant/__main__.py
View File

@ -20,7 +20,6 @@ except ImportError:
def main():
""" Starts Home Assistant. Will create demo config if no config found. """
tasks = ['serve', 'test']
parser = argparse.ArgumentParser()
parser.add_argument(
@ -29,64 +28,49 @@ def main():
default="config",
help="Directory that contains the Home Assistant configuration")
parser.add_argument(
'-t', '--task',
default=tasks[0],
choices=tasks,
help="Task to execute. Defaults to serve.")
args = parser.parse_args()
if args.task == tasks[1]:
# unittest does not like our command line arguments, remove them
sys.argv[1:] = []
# Validate that all core dependencies are installed
import_fail = False
import unittest
for module in ['requests']:
try:
importlib.import_module(module)
except ImportError:
import_fail = True
print(
'Fatal Error: Unable to find dependency {}'.format(module))
unittest.main(module='homeassistant.test')
if import_fail:
print(("Install dependencies by running: "
"pip3 install -r requirements.txt"))
exit()
else:
# Validate that all core dependencies are installed
import_fail = False
# Test if configuration directory exists
config_dir = os.path.join(os.getcwd(), args.config)
for module in ['requests']:
try:
importlib.import_module(module)
except ImportError:
import_fail = True
print(
'Fatal Error: Unable to find dependency {}'.format(module))
if not os.path.isdir(config_dir):
print(('Fatal Error: Unable to find specified configuration '
'directory {} ').format(config_dir))
sys.exit()
if import_fail:
print(("Install dependencies by running: "
"pip3 install -r requirements.txt"))
exit()
config_path = os.path.join(config_dir, 'home-assistant.conf')
# Test if configuration directory exists
config_dir = os.path.join(os.getcwd(), args.config)
if not os.path.isdir(config_dir):
print(('Fatal Error: Unable to find specified configuration '
'directory {} ').format(config_dir))
# Ensure a config file exists to make first time usage easier
if not os.path.isfile(config_path):
try:
with open(config_path, 'w') as conf:
conf.write("[http]\n")
conf.write("api_password=password\n\n")
conf.write("[demo]\n")
except IOError:
print(('Fatal Error: No configuration file found and unable '
'to write a default one to {}').format(config_path))
sys.exit()
config_path = os.path.join(config_dir, 'home-assistant.conf')
# Ensure a config file exists to make first time usage easier
if not os.path.isfile(config_path):
try:
with open(config_path, 'w') as conf:
conf.write("[http]\n")
conf.write("api_password=password\n\n")
conf.write("[demo]\n")
except IOError:
print(('Fatal Error: No configuration file found and unable '
'to write a default one to {}').format(config_path))
sys.exit()
hass = bootstrap.from_config_file(config_path)
hass.start()
hass.block_till_stopped()
hass = bootstrap.from_config_file(config_path)
hass.start()
hass.block_till_stopped()
if __name__ == "__main__":
main()

14
homeassistant/components/device_tracker/__init__.py
View File

@ -174,10 +174,9 @@ class DeviceTracker(object):
is_new_file = not os.path.isfile(known_dev_path)
with open(known_dev_path, 'a') as outp:
_LOGGER.info((
"Found {} new devices,"
" updating {}").format(len(unknown_devices),
known_dev_path))
_LOGGER.info(
"Found %d new devices, updating %s",
len(unknown_devices), known_dev_path)
writer = csv.writer(outp)
@ -197,10 +196,9 @@ class DeviceTracker(object):
'picture': ""}
except IOError:
_LOGGER.exception((
"Error updating {}"
"with {} new devices").format(known_dev_path,
len(unknown_devices)))
_LOGGER.exception(
"Error updating %s with %d new devices",
known_dev_path, len(unknown_devices))
self.lock.release()

10
homeassistant/components/http/__init__.py
View File

@ -110,7 +110,7 @@ CONF_DEVELOPMENT = "development"
_LOGGER = logging.getLogger(__name__)
# TODO add shutdown https://docs.python.org/3.4/library/socketserver.html#socketserver.BaseServer.shutdown
def setup(hass, config):
""" Sets up the HTTP API and debug interface. """
@ -136,20 +136,24 @@ def setup(hass, config):
lambda event:
threading.Thread(target=server.start, daemon=True).start())
hass.listen_once_event(
ha.EVENT_HOMEASSISTANT_STOP,
lambda event: server.shutdown())
# If no local api set, set one with known information
if isinstance(hass, rem.HomeAssistant) and hass.local_api is None:
hass.local_api = \
rem.API(util.get_local_ip(), api_password, server_port)
hass.server = server
return True
class HomeAssistantHTTPServer(ThreadingMixIn, HTTPServer):
""" Handle HTTP requests in a threaded fashion. """
# pylint: disable=too-few-public-methods
allow_reuse_address = True
daemon_threads = True
# pylint: disable=too-many-arguments
def __init__(self, server_address, RequestHandlerClass,

29
homeassistant/remote.py
View File

@ -18,7 +18,6 @@ import urllib.parse
import requests
import homeassistant as ha
from homeassistant.remote_json import JSONEncoder
SERVER_PORT = 8123
@ -39,9 +38,9 @@ METHOD_POST = "post"
_LOGGER = logging.getLogger(__name__)
# pylint: disable=no-init, invalid-name
class APIStatus(enum.Enum):
""" Represents API status. """
# pylint: disable=no-init,invalid-name,too-few-public-methods
OK = "ok"
INVALID_PASSWORD = "invalid_password"
@ -135,9 +134,22 @@ class HomeAssistant(ha.HomeAssistant):
self.bus.fire(ha.EVENT_HOMEASSISTANT_START,
origin=ha.EventOrigin.remote)
def stop(self):
""" Stops Home Assistant and shuts down all threads. """
_LOGGER.info("Stopping")
self.bus.fire(ha.EVENT_HOMEASSISTANT_STOP,
origin=ha.EventOrigin.remote)
# Wait till all responses to homeassistant_stop are done
self._pool.block_till_done()
self._pool.stop()
class EventBus(ha.EventBus):
""" EventBus implementation that forwards fire_event to remote API. """
# pylint: disable=too-few-public-methods
def __init__(self, api, pool=None):
super().__init__(pool)
@ -239,6 +251,19 @@ class StateMachine(ha.StateMachine):
self._states[event.data['entity_id']] = event.data['new_state']
class JSONEncoder(json.JSONEncoder):
""" JSONEncoder that supports Home Assistant objects. """
# pylint: disable=too-few-public-methods,method-hidden
def default(self, obj):
""" Converts Home Assistant objects and hands
other objects to the original method. """
if isinstance(obj, ha.State):
return obj.as_dict()
return json.JSONEncoder.default(self, obj)
def validate_api(api):
""" Makes a call to validate API. """
try:

21
homeassistant/remote_json.py
View File

@ -1,21 +0,0 @@
# pylint: skip-file
"""
Helper methods for using JSON.
This used to be in homeassistant.remote but has been moved
to this module because of a bug in PyLint that would make it crash.
"""
import homeassistant as ha
import json
class JSONEncoder(json.JSONEncoder):
""" JSONEncoder that supports Home Assistant objects. """
def default(self, obj):
""" Checks if Home Assistat object and encodes if possible.
Else hand it off to original method. """
if isinstance(obj, ha.State):
return obj.as_dict()
return json.JSONEncoder.default(self, obj)

16
homeassistant/util.py
View File

@ -153,7 +153,7 @@ def get_local_ip():
class OrderedEnum(enum.Enum):
""" Taken from Python 3.4.0 docs. """
# pylint: disable=no-init
# pylint: disable=no-init, too-few-public-methods
def __ge__(self, other):
if self.__class__ is other.__class__:
@ -215,8 +215,8 @@ class ThreadPool(object):
""" A simple queue-based thread pool.
Will initiate it's workers using worker(queue).start() """
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-few-public-methods
def __init__(self, worker_count, job_handler, busy_callback=None):
"""
worker_count: number of threads to run that handle jobs
@ -224,8 +224,8 @@ class ThreadPool(object):
busy_callback: method to be called when queue gets too big.
Parameters: list_of_current_jobs, number_pending_jobs
"""
work_queue = self.work_queue = queue.PriorityQueue()
current_jobs = self.current_jobs = []
self.work_queue = work_queue = queue.PriorityQueue()
self.current_jobs = current_jobs = []
self.worker_count = worker_count
self.busy_callback = busy_callback
self.busy_warning_limit = worker_count**2
@ -260,19 +260,21 @@ class ThreadPool(object):
def block_till_done(self):
""" Blocks till all work is done. """
with self._lock:
self.work_queue.join()
self.work_queue.join()
def stop(self):
""" Stops all the threads. """
with self._lock:
if not self.running:
return
# Clear the queue
while self.work_queue.qsize() > 0:
self.work_queue.get()
self.work_queue.task_done()
# Tell the workers to quit
for i in range(self.worker_count):
for _ in range(self.worker_count):
self.add_job(1000, self._quit_task)
self.running = False

5
run_tests.sh Executable file
View File

@ -0,0 +1,5 @@
#!/bin/bash
pylint homeassistant
flake8 homeassistant --exclude bower_components,external
python3 -m unittest discover test

11
test/components/__init__.py → test/test_component_core.py
View File

@ -1,9 +1,8 @@
"""
homeassistant.test
~~~~~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant modules do what they should do.
test.test_component_core
~~~~~~~~~~~~~~~~~~~~~~~~
Tests core compoments.
"""
# pylint: disable=protected-access,too-many-public-methods
import unittest
@ -25,6 +24,10 @@ class TestComponentsCore(unittest.TestCase):
self.hass.states.set('light.Bowl', comps.STATE_ON)
self.hass.states.set('light.Ceiling', comps.STATE_OFF)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass._pool.stop()
def test_is_on(self):
""" Test is_on method. """
self.assertTrue(comps.is_on(self.hass, 'light.Bowl'))

84
test/components/http.py → test/test_component_http.py
View File

@ -1,14 +1,14 @@
"""
homeassistant.test
~~~~~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant modules do what they should do.
test.test_component_http
~~~~~~~~~~~~~~~~~~~~~~~~
Tests Home Assistant HTTP component does what it should do.
"""
# pylint: disable=protected-access,too-many-public-methods
import re
import unittest
import json
import logging
import requests
@ -16,21 +16,49 @@ import homeassistant as ha
import homeassistant.remote as remote
import homeassistant.components.http as http
from test.remote import _url, ensure_homeassistant_started
API_PASSWORD = "test1234"
# Somehow the socket that holds the default port does not get released
# when we close down HA in a different test case. Until I have figured
# out what is going on, let's run this test on a different port.
SERVER_PORT = 8120
HTTP_BASE_URL = "http://127.0.0.1:{}".format(SERVER_PORT)
HA_HEADERS = {remote.AUTH_HEADER: API_PASSWORD}
def _url(path=""):
""" Helper method to generate urls. """
return HTTP_BASE_URL + path
def setUpModule():
""" Initalizes a Home Assistant server. """
global hass
hass = ha.HomeAssistant()
hass.bus.listen('test_event', lambda _: _)
hass.states.set('test.test', 'a_state')
http.setup(hass,
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: SERVER_PORT}})
hass.start()
def tearDownModule():
""" Stops the Home Assistant server. """
global hass
hass.stop()
class TestHTTP(unittest.TestCase):
""" Test the HTTP debug interface and API. """
@classmethod
def setUpClass(cls): # pylint: disable=invalid-name
""" things to be run when tests are started. """
cls.hass = ensure_homeassistant_started()
def test_get_frontend(self):
""" Tests if we can get the frontend. """
req = requests.get(_url(""))
@ -68,7 +96,7 @@ class TestHTTP(unittest.TestCase):
remote_data = [ha.State.from_dict(item) for item in req.json()]
self.assertEqual(self.hass.states.all(), remote_data)
self.assertEqual(hass.states.all(), remote_data)
def test_api_get_state(self):
""" Test if the debug interface allows us to get a state. """
@ -78,7 +106,7 @@ class TestHTTP(unittest.TestCase):
data = ha.State.from_dict(req.json())
state = self.hass.states.get("test.test")
state = hass.states.get("test.test")
self.assertEqual(state.state, data.state)
self.assertEqual(state.last_changed, data.last_changed)
@ -95,14 +123,14 @@ class TestHTTP(unittest.TestCase):
def test_api_state_change(self):
""" Test if we can change the state of an entity that exists. """
self.hass.states.set("test.test", "not_to_be_set")
hass.states.set("test.test", "not_to_be_set")
requests.post(_url(remote.URL_API_STATES_ENTITY.format("test.test")),
data=json.dumps({"state": "debug_state_change2",
"api_password": API_PASSWORD}))
self.assertEqual("debug_state_change2",
self.hass.states.get("test.test").state)
hass.states.get("test.test").state)
# pylint: disable=invalid-name
def test_api_state_change_of_non_existing_entity(self):
@ -117,7 +145,7 @@ class TestHTTP(unittest.TestCase):
data=json.dumps({"state": new_state,
"api_password": API_PASSWORD}))
cur_state = (self.hass.states.
cur_state = (hass.states.
get("test_entity.that_does_not_exist").state)
self.assertEqual(201, req.status_code)
@ -132,13 +160,13 @@ class TestHTTP(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
self.hass.listen_once_event("test.event_no_data", listener)
hass.listen_once_event("test.event_no_data", listener)
requests.post(
_url(remote.URL_API_EVENTS_EVENT.format("test.event_no_data")),
headers=HA_HEADERS)
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))
@ -153,14 +181,14 @@ class TestHTTP(unittest.TestCase):
if "test" in event.data:
test_value.append(1)
self.hass.listen_once_event("test_event_with_data", listener)
hass.listen_once_event("test_event_with_data", listener)
requests.post(
_url(remote.URL_API_EVENTS_EVENT.format("test_event_with_data")),
data=json.dumps({"test": 1}),
headers=HA_HEADERS)
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))
@ -173,14 +201,14 @@ class TestHTTP(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
self.hass.listen_once_event("test_event_bad_data", listener)
hass.listen_once_event("test_event_bad_data", listener)
req = requests.post(
_url(remote.URL_API_EVENTS_EVENT.format("test_event_bad_data")),
data=json.dumps('not an object'),
headers=HA_HEADERS)
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(422, req.status_code)
self.assertEqual(0, len(test_value))
@ -190,7 +218,7 @@ class TestHTTP(unittest.TestCase):
req = requests.get(_url(remote.URL_API_EVENTS),
headers=HA_HEADERS)
local = self.hass.bus.listeners
local = hass.bus.listeners
for event in req.json():
self.assertEqual(event["listener_count"],
@ -203,7 +231,7 @@ class TestHTTP(unittest.TestCase):
req = requests.get(_url(remote.URL_API_SERVICES),
headers=HA_HEADERS)
local_services = self.hass.services.services
local_services = hass.services.services
for serv_domain in req.json():
local = local_services.pop(serv_domain["domain"])
@ -218,14 +246,14 @@ class TestHTTP(unittest.TestCase):
""" Helper method that will verify that our service got called. """
test_value.append(1)
self.hass.services.register("test_domain", "test_service", listener)
hass.services.register("test_domain", "test_service", listener)
requests.post(
_url(remote.URL_API_SERVICES_SERVICE.format(
"test_domain", "test_service")),
headers=HA_HEADERS)
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))
@ -239,7 +267,7 @@ class TestHTTP(unittest.TestCase):
if "test" in service_call.data:
test_value.append(1)
self.hass.services.register("test_domain", "test_service", listener)
hass.services.register("test_domain", "test_service", listener)
requests.post(
_url(remote.URL_API_SERVICES_SERVICE.format(
@ -247,6 +275,6 @@ class TestHTTP(unittest.TestCase):
data=json.dumps({"test": 1}),
headers=HA_HEADERS)
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))

31
test/__init__.py → test/test_core.py
View File

@ -1,9 +1,8 @@
"""
homeassistant.test
~~~~~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant modules do what they should do.
test.test_core
~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant core works.
"""
# pylint: disable=protected-access,too-many-public-methods
import os
@ -28,6 +27,10 @@ class TestHomeAssistant(unittest.TestCase):
self.hass.states.set("light.Bowl", "on")
self.hass.states.set("switch.AC", "off")
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass._pool.stop()
def test_get_config_path(self):
""" Test get_config_path method. """
self.assertEqual(os.getcwd(), self.hass.config_dir)
@ -43,7 +46,7 @@ class TestHomeAssistant(unittest.TestCase):
blocking_thread.start()
# Python will now give attention to the other thread
time.sleep(.01)
time.sleep(1)
self.assertTrue(blocking_thread.is_alive())
@ -205,6 +208,10 @@ class TestEventBus(unittest.TestCase):
self.bus = ha.EventBus()
self.bus.listen('test_event', lambda x: len)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.bus._pool.stop()
def test_add_remove_listener(self):
""" Test remove_listener method. """
old_count = len(self.bus.listeners)
@ -257,6 +264,10 @@ class TestStateMachine(unittest.TestCase):
self.states.set("light.Bowl", "on")
self.states.set("switch.AC", "off")
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.bus._pool.stop()
def test_is_state(self):
""" Test is_state method. """
self.assertTrue(self.states.is_state('light.Bowl', 'on'))
@ -291,11 +302,15 @@ class TestServiceRegistry(unittest.TestCase):
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.pool = ha.create_worker_pool()
self.bus = ha.EventBus(self.pool)
self.services = ha.ServiceRegistry(self.bus, self.pool)
self.services.register("test_domain", "test_service", lambda x: len)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.pool.stop()
def test_has_service(self):
""" Test has_service method. """
self.assertTrue(

9
test/loader.py → test/test_loader.py
View File

@ -1,9 +1,8 @@
"""
homeassistant.test
test.test_loader
~~~~~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant modules do what they should do.
Provides tests to verify that we can load components.
"""
# pylint: disable=too-many-public-methods
import unittest
@ -19,6 +18,10 @@ class TestLoader(unittest.TestCase):
self.hass = ha.HomeAssistant()
loader.prepare(self.hass)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass._pool.stop()
def test_get_component(self):
""" Test if get_component works. """
self.assertEqual(http, loader.get_component('http'))

153
test/remote.py → test/test_remote.py
View File

@ -1,12 +1,12 @@
"""
homeassistant.test
~~~~~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant modules do what they should do.
test.remote
~~~~~~~~~~~
Tests Home Assistant remote methods and classes.
"""
# pylint: disable=protected-access,too-many-public-methods
import unittest
import logging
import homeassistant as ha
import homeassistant.remote as remote
@ -24,66 +24,47 @@ def _url(path=""):
return HTTP_BASE_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 setUpModule():
""" Initalizes a Home Assistant server and Slave instance. """
global hass, slave, master_api
hass = ha.HomeAssistant()
hass.bus.listen('test_event', lambda _: _)
hass.states.set('test.test', 'a_state')
http.setup(hass,
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD}})
hass.start()
master_api = remote.API("127.0.0.1", API_PASSWORD)
# Start slave
local_api = remote.API("127.0.0.1", API_PASSWORD, 8124)
slave = remote.HomeAssistant(master_api, local_api)
http.setup(slave,
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: 8124}})
slave.start()
def ensure_homeassistant_started():
""" Ensures home assistant is started. """
def tearDownModule():
""" Stops the Home Assistant server and slave. """
global hass, slave
if not HAHelper.hass:
print("Setting up new HA")
hass = ha.HomeAssistant()
hass.bus.listen('test_event', lambda _: _)
hass.states.set('test.test', 'a_state')
http.setup(hass,
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD}})
hass.start()
HAHelper.hass = hass
return HAHelper.hass
def ensure_slave_started():
""" Ensure a home assistant slave is started. """
ensure_homeassistant_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(remote_api, local_api)
http.setup(slave,
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: 8124}})
slave.start()
HAHelper.slave = slave
return HAHelper.slave
hass.stop()
slave.stop()
class TestRemoteMethods(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.api = remote.API("127.0.0.1", API_PASSWORD)
def test_validate_api(self):
""" Test Python API validate_api. """
self.assertEqual(remote.APIStatus.OK, remote.validate_api(self.api))
self.assertEqual(remote.APIStatus.OK, remote.validate_api(master_api))
self.assertEqual(remote.APIStatus.INVALID_PASSWORD,
remote.validate_api(
@ -91,8 +72,8 @@ class TestRemoteMethods(unittest.TestCase):
def test_get_event_listeners(self):
""" Test Python API get_event_listeners. """
local_data = self.hass.bus.listeners
remote_data = remote.get_event_listeners(self.api)
local_data = hass.bus.listeners
remote_data = remote.get_event_listeners(master_api)
for event in remote_data:
self.assertEqual(local_data.pop(event["event"]),
@ -108,11 +89,11 @@ class TestRemoteMethods(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
self.hass.listen_once_event("test.event_no_data", listener)
hass.listen_once_event("test.event_no_data", listener)
remote.fire_event(self.api, "test.event_no_data")
remote.fire_event(master_api, "test.event_no_data")
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))
@ -120,34 +101,34 @@ class TestRemoteMethods(unittest.TestCase):
""" Test Python API get_state. """
self.assertEqual(
self.hass.states.get('test.test'),
remote.get_state(self.api, 'test.test'))
hass.states.get('test.test'),
remote.get_state(master_api, 'test.test'))
def test_get_states(self):
""" Test Python API get_state_entity_ids. """
self.assertEqual(
remote.get_states(self.api), self.hass.states.all())
remote.get_states(master_api), hass.states.all())
def test_set_state(self):
""" Test Python API set_state. """
self.assertTrue(remote.set_state(self.api, 'test.test', 'set_test'))
self.assertTrue(remote.set_state(master_api, 'test.test', 'set_test'))
self.assertEqual('set_test', self.hass.states.get('test.test').state)
self.assertEqual('set_test', hass.states.get('test.test').state)
def test_is_state(self):
""" Test Python API is_state. """
self.assertTrue(
remote.is_state(self.api, 'test.test',
self.hass.states.get('test.test').state))
remote.is_state(master_api, 'test.test',
hass.states.get('test.test').state))
def test_get_services(self):
""" Test Python API get_services. """
local_services = self.hass.services.services
local_services = hass.services.services
for serv_domain in remote.get_services(self.api):
for serv_domain in remote.get_services(master_api):
local = local_services.pop(serv_domain["domain"])
self.assertEqual(local, serv_domain["services"])
@ -160,11 +141,11 @@ class TestRemoteMethods(unittest.TestCase):
""" Helper method that will verify that our service got called. """
test_value.append(1)
self.hass.services.register("test_domain", "test_service", listener)
hass.services.register("test_domain", "test_service", listener)
remote.call_service(self.api, "test_domain", "test_service")
remote.call_service(master_api, "test_domain", "test_service")
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))
@ -172,12 +153,6 @@ class TestRemoteMethods(unittest.TestCase):
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_home_assistant_init(self):
""" Test HomeAssistant init. """
self.assertRaises(
@ -186,24 +161,24 @@ 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()))
self.assertEqual(len(hass.states.all()),
len(slave.states.all()))
for state in self.hass.states.all():
for state in hass.states.all():
self.assertEqual(
state, self.slave.states.get(state.entity_id))
state, slave.states.get(state.entity_id))
def test_statemachine_set(self):
""" Tests if setting the state on a slave is recorded. """
self.slave.states.set("remote.test", "remote.statemachine test")
slave.states.set("remote.test", "remote.statemachine test")
# Wait till slave tells master
self.slave._pool.block_till_done()
slave._pool.block_till_done()
# Wait till master gives updated state
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual("remote.statemachine test",
self.slave.states.get("remote.test").state)
slave.states.get("remote.test").state)
def test_eventbus_fire(self):
""" Test if events fired from the eventbus get fired. """
@ -213,13 +188,13 @@ class TestRemoteClasses(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
self.slave.listen_once_event("test.event_no_data", listener)
slave.listen_once_event("test.event_no_data", listener)
self.slave.bus.fire("test.event_no_data")
slave.bus.fire("test.event_no_data")
# Wait till slave tells master
self.slave._pool.block_till_done()
slave._pool.block_till_done()
# Wait till master gives updated event
self.hass._pool.block_till_done()
hass._pool.block_till_done()
self.assertEqual(1, len(test_value))

7
test/util.py → test/test_util.py
View File

@ -1,9 +1,8 @@
"""
homeassistant.test
~~~~~~~~~~~~~~~~~~
Provides tests to verify that Home Assistant modules do what they should do.
test.test_util
~~~~~~~~~~~~~~
Tests Home Assistant util methods.
"""
# pylint: disable=too-many-public-methods
import unittest