From b90964faad64b2b532988d660154496324a0a5a6 Mon Sep 17 00:00:00 2001 From: Martin Berg Date: Mon, 5 Jun 2017 08:53:25 +0200 Subject: [PATCH] Add support for Vanderbilt SPC alarm panels and attached sensors (#7663) * Add support for Vanderbilt SPC alarm panels. * Arm/disarm + read state * Autodiscover and add motion sensors * Fix code formatting. * Use asyncio.async for Python < 3.4.4. * Fix for moved aiohttp exceptions. * Add docstrings. * Fix tests and add docstrings. --- .../components/alarm_control_panel/spc.py | 96 ++++++ homeassistant/components/binary_sensor/spc.py | 99 +++++++ homeassistant/components/spc.py | 279 ++++++++++++++++++ requirements_all.txt | 1 + .../alarm_control_panel/test_spc.py | 64 ++++ tests/components/binary_sensor/test_spc.py | 67 +++++ tests/components/test_spc.py | 149 ++++++++++ 7 files changed, 755 insertions(+) create mode 100644 homeassistant/components/alarm_control_panel/spc.py create mode 100644 homeassistant/components/binary_sensor/spc.py create mode 100644 homeassistant/components/spc.py create mode 100644 tests/components/alarm_control_panel/test_spc.py create mode 100644 tests/components/binary_sensor/test_spc.py create mode 100644 tests/components/test_spc.py diff --git a/homeassistant/components/alarm_control_panel/spc.py b/homeassistant/components/alarm_control_panel/spc.py new file mode 100644 index 00000000000..de4d5098b41 --- /dev/null +++ b/homeassistant/components/alarm_control_panel/spc.py @@ -0,0 +1,96 @@ +""" +Support for Vanderbilt (formerly Siemens) SPC alarm systems. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/alarm_control_panel.spc/ +""" +import asyncio +import logging + +import homeassistant.components.alarm_control_panel as alarm +from homeassistant.components.spc import ( + SpcWebGateway, ATTR_DISCOVER_AREAS, DATA_API, DATA_REGISTRY) +from homeassistant.const import ( + STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED, + STATE_UNKNOWN) + + +_LOGGER = logging.getLogger(__name__) + +SPC_AREA_MODE_TO_STATE = {'0': STATE_ALARM_DISARMED, + '1': STATE_ALARM_ARMED_HOME, + '3': STATE_ALARM_ARMED_AWAY} + + +def _get_alarm_state(spc_mode): + return SPC_AREA_MODE_TO_STATE.get(spc_mode, STATE_UNKNOWN) + + +@asyncio.coroutine +def async_setup_platform(hass, config, async_add_entities, + discovery_info=None): + """Set up the SPC alarm control panel platform.""" + if (discovery_info is None or + discovery_info[ATTR_DISCOVER_AREAS] is None): + return + + entities = [SpcAlarm(hass=hass, + area_id=area['id'], + name=area['name'], + state=_get_alarm_state(area['mode'])) + for area in discovery_info[ATTR_DISCOVER_AREAS]] + + async_add_entities(entities) + + +class SpcAlarm(alarm.AlarmControlPanel): + """Represents the SPC alarm panel.""" + + def __init__(self, hass, area_id, name, state): + """Initialize the SPC alarm panel.""" + self._hass = hass + self._area_id = area_id + self._name = name + self._state = state + self._api = hass.data[DATA_API] + + hass.data[DATA_REGISTRY].register_alarm_device(area_id, self) + + @asyncio.coroutine + def async_update_from_spc(self, state): + """Update the alarm panel with a new state.""" + self._state = state + yield from self.async_update_ha_state() + + @property + def should_poll(self): + """No polling needed.""" + return False + + @property + def name(self): + """Return the name of the device.""" + return self._name + + @property + def state(self): + """Return the state of the device.""" + return self._state + + @asyncio.coroutine + def async_alarm_disarm(self, code=None): + """Send disarm command.""" + yield from self._api.send_area_command( + self._area_id, SpcWebGateway.AREA_COMMAND_UNSET) + + @asyncio.coroutine + def async_alarm_arm_home(self, code=None): + """Send arm home command.""" + yield from self._api.send_area_command( + self._area_id, SpcWebGateway.AREA_COMMAND_PART_SET) + + @asyncio.coroutine + def async_alarm_arm_away(self, code=None): + """Send arm away command.""" + yield from self._api.send_area_command( + self._area_id, SpcWebGateway.AREA_COMMAND_SET) diff --git a/homeassistant/components/binary_sensor/spc.py b/homeassistant/components/binary_sensor/spc.py new file mode 100644 index 00000000000..8023e1cf4b3 --- /dev/null +++ b/homeassistant/components/binary_sensor/spc.py @@ -0,0 +1,99 @@ +""" +Support for Vanderbilt (formerly Siemens) SPC alarm systems. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/binary_sensor.spc/ +""" +import logging +import asyncio + +from homeassistant.components.spc import ( + ATTR_DISCOVER_DEVICES, DATA_REGISTRY) +from homeassistant.components.binary_sensor import BinarySensorDevice +from homeassistant.const import (STATE_UNAVAILABLE, STATE_ON, STATE_OFF) + + +_LOGGER = logging.getLogger(__name__) + +SPC_TYPE_TO_DEVICE_CLASS = {'0': 'motion', + '1': 'opening', + '3': 'smoke'} + + +SPC_INPUT_TO_SENSOR_STATE = {'0': STATE_OFF, + '1': STATE_ON} + + +def _get_device_class(spc_type): + return SPC_TYPE_TO_DEVICE_CLASS.get(spc_type, None) + + +def _get_sensor_state(spc_input): + return SPC_INPUT_TO_SENSOR_STATE.get(spc_input, STATE_UNAVAILABLE) + + +def _create_sensor(hass, zone): + return SpcBinarySensor(zone_id=zone['id'], + name=zone['zone_name'], + state=_get_sensor_state(zone['input']), + device_class=_get_device_class(zone['type']), + spc_registry=hass.data[DATA_REGISTRY]) + + +@asyncio.coroutine +def async_setup_platform(hass, config, async_add_entities, + discovery_info=None): + """Initialize the platform.""" + if (discovery_info is None or + discovery_info[ATTR_DISCOVER_DEVICES] is None): + return + + async_add_entities( + _create_sensor(hass, zone) + for zone in discovery_info[ATTR_DISCOVER_DEVICES] + if _get_device_class(zone['type'])) + + +class SpcBinarySensor(BinarySensorDevice): + """Represents a sensor based on an SPC zone.""" + + def __init__(self, zone_id, name, state, device_class, spc_registry): + """Initialize the sensor device.""" + self._zone_id = zone_id + self._name = name + self._state = state + self._device_class = device_class + + spc_registry.register_sensor_device(zone_id, self) + + @asyncio.coroutine + def async_update_from_spc(self, state): + """Update the state of the device.""" + self._state = state + yield from self.async_update_ha_state() + + @property + def name(self): + """The name of the device.""" + return self._name + + @property + def is_on(self): + """Whether the device is switched on.""" + return self._state == STATE_ON + + @property + def hidden(self) -> bool: + """Whether the device is hidden by default.""" + # these type of sensors are probably mainly used for automations + return True + + @property + def should_poll(self): + """No polling needed.""" + return False + + @property + def device_class(self): + """The device class.""" + return self._device_class diff --git a/homeassistant/components/spc.py b/homeassistant/components/spc.py new file mode 100644 index 00000000000..a271297d0fd --- /dev/null +++ b/homeassistant/components/spc.py @@ -0,0 +1,279 @@ +""" +Support for Vanderbilt (formerly Siemens) SPC alarm systems. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/spc/ +""" +import logging +import asyncio +import json +from urllib.parse import urljoin + +import aiohttp +import async_timeout +import voluptuous as vol + +import homeassistant.helpers.config_validation as cv +from homeassistant.helpers import discovery +from homeassistant.const import ( + STATE_UNKNOWN, STATE_ON, STATE_OFF, STATE_ALARM_ARMED_AWAY, + STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED, + STATE_UNAVAILABLE) + +DOMAIN = 'spc' +REQUIREMENTS = ['websockets==3.2'] + +_LOGGER = logging.getLogger(__name__) + +ATTR_DISCOVER_DEVICES = 'devices' +ATTR_DISCOVER_AREAS = 'areas' + +CONF_WS_URL = 'ws_url' +CONF_API_URL = 'api_url' + +DATA_REGISTRY = 'spc_registry' +DATA_API = 'spc_api' + +CONFIG_SCHEMA = vol.Schema({ + DOMAIN: vol.Schema({ + vol.Required(CONF_WS_URL): cv.string, + vol.Required(CONF_API_URL): cv.string + }), +}, extra=vol.ALLOW_EXTRA) + + +@asyncio.coroutine +def async_setup(hass, config): + """Setup the SPC platform.""" + hass.data[DATA_REGISTRY] = SpcRegistry() + + api = SpcWebGateway(hass, + config[DOMAIN].get(CONF_API_URL), + config[DOMAIN].get(CONF_WS_URL)) + + hass.data[DATA_API] = api + + # add sensor devices for each zone (typically motion/fire/door sensors) + zones = yield from api.get_zones() + if zones: + hass.async_add_job(discovery.async_load_platform( + hass, 'binary_sensor', DOMAIN, + {ATTR_DISCOVER_DEVICES: zones}, config)) + + # create a separate alarm panel for each area + areas = yield from api.get_areas() + if areas: + hass.async_add_job(discovery.async_load_platform( + hass, 'alarm_control_panel', DOMAIN, + {ATTR_DISCOVER_AREAS: areas}, config)) + + # start listening for incoming events over websocket + api.start_listener(_async_process_message, hass.data[DATA_REGISTRY]) + + return True + + +@asyncio.coroutine +def _async_process_message(sia_message, spc_registry): + spc_id = sia_message['sia_address'] + sia_code = sia_message['sia_code'] + + # BA - Burglary Alarm + # CG - Close Area + # NL - Perimeter Armed + # OG - Open Area + # ZO - Zone Open + # ZC - Zone Close + # ZX - Zone Short + # ZD - Zone Disconnected + + if sia_code in ('BA', 'CG', 'NL', 'OG', 'OQ'): + # change in area status, notify alarm panel device + device = spc_registry.get_alarm_device(spc_id) + else: + # change in zone status, notify sensor device + device = spc_registry.get_sensor_device(spc_id) + + sia_code_to_state_map = {'BA': STATE_ALARM_TRIGGERED, + 'CG': STATE_ALARM_ARMED_AWAY, + 'NL': STATE_ALARM_ARMED_HOME, + 'OG': STATE_ALARM_DISARMED, + 'OQ': STATE_ALARM_DISARMED, + 'ZO': STATE_ON, + 'ZC': STATE_OFF, + 'ZX': STATE_UNKNOWN, + 'ZD': STATE_UNAVAILABLE} + + new_state = sia_code_to_state_map.get(sia_code, None) + + if new_state and not device: + _LOGGER.warning("No device mapping found for SPC area/zone id %s.", + spc_id) + elif new_state: + yield from device.async_update_from_spc(new_state) + + +class SpcRegistry: + """Maintains mappings between SPC zones/areas and HA entities.""" + + def __init__(self): + """Initialize the registry.""" + self._zone_id_to_sensor_map = {} + self._area_id_to_alarm_map = {} + + def register_sensor_device(self, zone_id, device): + """Add a sensor device to the registry.""" + self._zone_id_to_sensor_map[zone_id] = device + + def get_sensor_device(self, zone_id): + """Retrieve a sensor device for a specific zone.""" + return self._zone_id_to_sensor_map.get(zone_id, None) + + def register_alarm_device(self, area_id, device): + """Add an alarm device to the registry.""" + self._area_id_to_alarm_map[area_id] = device + + def get_alarm_device(self, area_id): + """Retrieve an alarm device for a specific area.""" + return self._area_id_to_alarm_map.get(area_id, None) + + +@asyncio.coroutine +def _ws_process_message(message, async_callback, *args): + if message.get('status', '') != 'success': + _LOGGER.warning("Unsuccessful websocket message " + "delivered, ignoring: %s", message) + try: + yield from async_callback(message['data']['sia'], *args) + except: # pylint: disable=bare-except + _LOGGER.exception("Exception in callback, ignoring.") + + +class SpcWebGateway: + """Simple binding for the Lundix SPC Web Gateway REST API.""" + + AREA_COMMAND_SET = 'set' + AREA_COMMAND_PART_SET = 'set_a' + AREA_COMMAND_UNSET = 'unset' + + def __init__(self, hass, api_url, ws_url): + """Initialize the web gateway client.""" + self._hass = hass + self._api_url = api_url + self._ws_url = ws_url + self._ws = None + + @asyncio.coroutine + def get_zones(self): + """Retrieve all available zones.""" + return (yield from self._get_data('zone')) + + @asyncio.coroutine + def get_areas(self): + """Retrieve all available areas.""" + return (yield from self._get_data('area')) + + @asyncio.coroutine + def send_area_command(self, area_id, command): + """Send an area command.""" + _LOGGER.debug("Sending SPC area command '%s' to area %s.", + command, area_id) + resource = "area/{}/{}".format(area_id, command) + return (yield from self._call_web_gateway(resource, use_get=False)) + + def start_listener(self, async_callback, *args): + """Start the websocket listener.""" + try: + from asyncio import ensure_future + except ImportError: + from asyncio import async as ensure_future + + ensure_future(self._ws_listen(async_callback, *args)) + + def _build_url(self, resource): + return urljoin(self._api_url, "spc/{}".format(resource)) + + @asyncio.coroutine + def _get_data(self, resource): + data = yield from self._call_web_gateway(resource) + if not data: + return False + if data['status'] != 'success': + _LOGGER.error("SPC Web Gateway call unsuccessful " + "for resource '%s'.", resource) + return False + return [item for item in data['data'][resource]] + + @asyncio.coroutine + def _call_web_gateway(self, resource, use_get=True): + response = None + session = None + url = self._build_url(resource) + try: + _LOGGER.debug("Attempting to retrieve SPC data from %s.", url) + session = aiohttp.ClientSession() + with async_timeout.timeout(10, loop=self._hass.loop): + action = session.get if use_get else session.put + response = yield from action(url) + if response.status != 200: + _LOGGER.error("SPC Web Gateway returned http " + "status %d, response %s.", + response.status, (yield from response.text())) + return False + result = yield from response.json() + except asyncio.TimeoutError: + _LOGGER.error("Timeout getting SPC data from %s.", url) + return False + except aiohttp.ClientError: + _LOGGER.exception("Error getting SPC data from %s.", url) + return False + finally: + if session: + yield from session.close() + if response: + yield from response.release() + _LOGGER.debug("Data from SPC: %s", result) + return result + + @asyncio.coroutine + def _ws_read(self): + import websockets as wslib + + try: + if not self._ws: + self._ws = yield from wslib.connect(self._ws_url) + _LOGGER.info("Connected to websocket at %s.", self._ws_url) + except Exception as ws_exc: # pylint: disable=broad-except + _LOGGER.error("Failed to connect to websocket: %s", ws_exc) + return + + result = None + + try: + result = yield from self._ws.recv() + _LOGGER.debug("Data from websocket: %s", result) + except Exception as ws_exc: # pylint: disable=broad-except + _LOGGER.error("Failed to read from websocket: %s", ws_exc) + try: + yield from self._ws.close() + finally: + self._ws = None + + return result + + @asyncio.coroutine + def _ws_listen(self, async_callback, *args): + try: + while True: + result = yield from self._ws_read() + + if result: + yield from _ws_process_message(json.loads(result), + async_callback, *args) + else: + _LOGGER.info("Trying again in 30 seconds.") + yield from asyncio.sleep(30) + + finally: + if self._ws: + yield from self._ws.close() diff --git a/requirements_all.txt b/requirements_all.txt index 994f9974855..5372042bf94 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -873,6 +873,7 @@ wakeonlan==0.2.2 # homeassistant.components.media_player.gpmdp websocket-client==0.37.0 +# homeassistant.components.spc # homeassistant.components.media_player.webostv websockets==3.2 diff --git a/tests/components/alarm_control_panel/test_spc.py b/tests/components/alarm_control_panel/test_spc.py new file mode 100644 index 00000000000..9fcfbfd56d2 --- /dev/null +++ b/tests/components/alarm_control_panel/test_spc.py @@ -0,0 +1,64 @@ +"""Tests for Vanderbilt SPC alarm control panel platform.""" +import asyncio + +import pytest + +from homeassistant.components.spc import SpcRegistry +from homeassistant.components.alarm_control_panel import spc +from tests.common import async_test_home_assistant +from homeassistant.const import ( + STATE_ALARM_ARMED_AWAY, STATE_ALARM_DISARMED) + + +@pytest.fixture +def hass(loop): + """Home Assistant fixture with device mapping registry.""" + hass = loop.run_until_complete(async_test_home_assistant(loop)) + hass.data['spc_registry'] = SpcRegistry() + hass.data['spc_api'] = None + yield hass + loop.run_until_complete(hass.async_stop()) + + +@asyncio.coroutine +def test_setup_platform(hass): + """Test adding areas as separate alarm control panel devices.""" + added_entities = [] + + def add_entities(entities): + nonlocal added_entities + added_entities = list(entities) + + areas = {'areas': [{ + 'id': '1', + 'name': 'House', + 'mode': '3', + 'last_set_time': '1485759851', + 'last_set_user_id': '1', + 'last_set_user_name': 'Pelle', + 'last_unset_time': '1485800564', + 'last_unset_user_id': '1', + 'last_unset_user_name': 'Pelle', + 'last_alarm': '1478174896' + }, { + 'id': '3', + 'name': 'Garage', + 'mode': '0', + 'last_set_time': '1483705803', + 'last_set_user_id': '9998', + 'last_set_user_name': 'Lisa', + 'last_unset_time': '1483705808', + 'last_unset_user_id': '9998', + 'last_unset_user_name': 'Lisa' + }]} + + yield from spc.async_setup_platform(hass=hass, + config={}, + async_add_entities=add_entities, + discovery_info=areas) + + assert len(added_entities) == 2 + assert added_entities[0].name == 'House' + assert added_entities[0].state == STATE_ALARM_ARMED_AWAY + assert added_entities[1].name == 'Garage' + assert added_entities[1].state == STATE_ALARM_DISARMED diff --git a/tests/components/binary_sensor/test_spc.py b/tests/components/binary_sensor/test_spc.py new file mode 100644 index 00000000000..2acd093dc1f --- /dev/null +++ b/tests/components/binary_sensor/test_spc.py @@ -0,0 +1,67 @@ +"""Tests for Vanderbilt SPC binary sensor platform.""" +import asyncio + +import pytest + +from homeassistant.components.spc import SpcRegistry +from homeassistant.components.binary_sensor import spc +from tests.common import async_test_home_assistant + + +@pytest.fixture +def hass(loop): + """Home Assistant fixture with device mapping registry.""" + hass = loop.run_until_complete(async_test_home_assistant(loop)) + hass.data['spc_registry'] = SpcRegistry() + yield hass + loop.run_until_complete(hass.async_stop()) + + +@asyncio.coroutine +def test_setup_platform(hass): + """Test autodiscovery of supported device types.""" + added_entities = [] + + zones = {'devices': [{ + 'id': '1', + 'type': '3', + 'zone_name': 'Kitchen smoke', + 'area': '1', + 'area_name': 'House', + 'input': '0', + 'status': '0', + }, { + 'id': '3', + 'type': '0', + 'zone_name': 'Hallway PIR', + 'area': '1', + 'area_name': 'House', + 'input': '0', + 'status': '0', + }, { + 'id': '5', + 'type': '1', + 'zone_name': 'Front door', + 'area': '1', + 'area_name': 'House', + 'input': '1', + 'status': '0', + }]} + + def add_entities(entities): + nonlocal added_entities + added_entities = list(entities) + + yield from spc.async_setup_platform(hass=hass, + config={}, + async_add_entities=add_entities, + discovery_info=zones) + + assert len(added_entities) == 3 + assert added_entities[0].device_class == 'smoke' + assert added_entities[0].state == 'off' + assert added_entities[1].device_class == 'motion' + assert added_entities[1].state == 'off' + assert added_entities[2].device_class == 'opening' + assert added_entities[2].state == 'on' + assert all(d.hidden for d in added_entities) diff --git a/tests/components/test_spc.py b/tests/components/test_spc.py new file mode 100644 index 00000000000..6fae8d821c2 --- /dev/null +++ b/tests/components/test_spc.py @@ -0,0 +1,149 @@ +"""Tests for Vanderbilt SPC component.""" +import asyncio + +import pytest + +from homeassistant.components import spc +from homeassistant.bootstrap import async_setup_component +from tests.common import async_test_home_assistant +from tests.test_util.aiohttp import mock_aiohttp_client +from homeassistant.const import (STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED) + + +@pytest.fixture +def hass(loop): + """Home Assistant fixture with device mapping registry.""" + hass = loop.run_until_complete(async_test_home_assistant(loop)) + hass.data[spc.DATA_REGISTRY] = spc.SpcRegistry() + hass.data[spc.DATA_API] = None + yield hass + loop.run_until_complete(hass.async_stop()) + + +@pytest.fixture +def spcwebgw(hass): + """Fixture for the SPC Web Gateway API configured for localhost.""" + yield spc.SpcWebGateway(hass=hass, + api_url='http://localhost/', + ws_url='ws://localhost/') + + +@pytest.fixture +def aioclient_mock(): + """HTTP client mock for areas and zones.""" + areas = """{"status":"success","data":{"area":[{"id":"1","name":"House", + "mode":"0","last_set_time":"1485759851","last_set_user_id":"1", + "last_set_user_name":"Pelle","last_unset_time":"1485800564", + "last_unset_user_id":"1","last_unset_user_name":"Pelle","last_alarm": + "1478174896"},{"id":"3","name":"Garage","mode":"0","last_set_time": + "1483705803","last_set_user_id":"9998","last_set_user_name":"Lisa", + "last_unset_time":"1483705808","last_unset_user_id":"9998", + "last_unset_user_name":"Lisa"}]}}""" + + zones = """{"status":"success","data":{"zone":[{"id":"1","type":"3", + "zone_name":"Kitchen smoke","area":"1","area_name":"House","input":"0", + "logic_input":"0","status":"0","proc_state":"0","inhibit_allowed":"1", + "isolate_allowed":"1"},{"id":"3","type":"0","zone_name":"Hallway PIR", + "area":"1","area_name":"House","input":"0","logic_input":"0","status": + "0","proc_state":"0","inhibit_allowed":"1","isolate_allowed":"1"}, + {"id":"5","type":"1","zone_name":"Front door","area":"1","area_name": + "House","input":"1","logic_input":"0","status":"0","proc_state":"0", + "inhibit_allowed":"1","isolate_allowed":"1"}]}}""" + + with mock_aiohttp_client() as mock_session: + mock_session.get('http://localhost/spc/area', text=areas) + mock_session.get('http://localhost/spc/zone', text=zones) + yield mock_session + + +@asyncio.coroutine +def test_update_alarm_device(hass, aioclient_mock, monkeypatch): + """Test that alarm panel state changes on incoming websocket data.""" + monkeypatch.setattr("homeassistant.components.spc.SpcWebGateway." + "start_listener", lambda x, *args: None) + config = { + 'spc': { + 'api_url': 'http://localhost/', + 'ws_url': 'ws://localhost/' + } + } + yield from async_setup_component(hass, 'spc', config) + yield from hass.async_block_till_done() + + entity_id = 'alarm_control_panel.house' + + assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED + + msg = {"sia_code": "NL", "sia_address": "1", "description": "House|Sam|1"} + yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY]) + assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME + + msg = {"sia_code": "OQ", "sia_address": "1", "description": "Sam"} + yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY]) + assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED + + +@asyncio.coroutine +def test_update_sensor_device(hass, aioclient_mock, monkeypatch): + """Test that sensors change state on incoming websocket data.""" + monkeypatch.setattr("homeassistant.components.spc.SpcWebGateway." + "start_listener", lambda x, *args: None) + config = { + 'spc': { + 'api_url': 'http://localhost/', + 'ws_url': 'ws://localhost/' + } + } + yield from async_setup_component(hass, 'spc', config) + yield from hass.async_block_till_done() + + assert hass.states.get('binary_sensor.hallway_pir').state == 'off' + + msg = {"sia_code": "ZO", "sia_address": "3", "description": "Hallway PIR"} + yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY]) + assert hass.states.get('binary_sensor.hallway_pir').state == 'on' + + msg = {"sia_code": "ZC", "sia_address": "3", "description": "Hallway PIR"} + yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY]) + assert hass.states.get('binary_sensor.hallway_pir').state == 'off' + + +class TestSpcRegistry: + """Test the device mapping registry.""" + + def test_sensor_device(self): + """Test retrieving device based on ID.""" + r = spc.SpcRegistry() + r.register_sensor_device('1', 'dummy') + assert r.get_sensor_device('1') == 'dummy' + + def test_alarm_device(self): + """Test retrieving device based on zone name.""" + r = spc.SpcRegistry() + r.register_alarm_device('Area 51', 'dummy') + assert r.get_alarm_device('Area 51') == 'dummy' + + +class TestSpcWebGateway: + """Test the SPC Web Gateway API wrapper.""" + + @asyncio.coroutine + def test_get_areas(self, spcwebgw, aioclient_mock): + """Test area retrieval.""" + result = yield from spcwebgw.get_areas() + assert aioclient_mock.call_count == 1 + assert len(list(result)) == 2 + + @asyncio.coroutine + @pytest.mark.parametrize("url_command,command", [ + ('set', spc.SpcWebGateway.AREA_COMMAND_SET), + ('unset', spc.SpcWebGateway.AREA_COMMAND_UNSET), + ('set_a', spc.SpcWebGateway.AREA_COMMAND_PART_SET) + ]) + def test_area_commands(self, spcwebgw, url_command, command): + """Test alarm arming/disarming.""" + with mock_aiohttp_client() as aioclient_mock: + url = "http://localhost/spc/area/1/{}".format(url_command) + aioclient_mock.put(url, text='{}') + yield from spcwebgw.send_area_command('1', command) + assert aioclient_mock.call_count == 1