diff --git a/homeassistant/components/alarm_control_panel/spc.py b/homeassistant/components/alarm_control_panel/spc.py index 2aa157a5cad..9150518022f 100644 --- a/homeassistant/components/alarm_control_panel/spc.py +++ b/homeassistant/components/alarm_control_panel/spc.py @@ -4,71 +4,65 @@ 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.helpers.dispatcher import async_dispatcher_connect +from homeassistant.core import callback from homeassistant.components.spc import ( - ATTR_DISCOVER_AREAS, DATA_API, DATA_REGISTRY, SpcWebGateway) + ATTR_DISCOVER_AREAS, DATA_API, SIGNAL_UPDATE_ALARM) from homeassistant.const import ( - STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED, - STATE_UNKNOWN) + STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT, + STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED) _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): +def _get_alarm_state(area): """Get the alarm state.""" - return SPC_AREA_MODE_TO_STATE.get(spc_mode, STATE_UNKNOWN) + from pyspcwebgw.const import AreaMode + + if area.verified_alarm: + return STATE_ALARM_TRIGGERED + + mode_to_state = { + AreaMode.UNSET: STATE_ALARM_DISARMED, + AreaMode.PART_SET_A: STATE_ALARM_ARMED_HOME, + AreaMode.PART_SET_B: STATE_ALARM_ARMED_NIGHT, + AreaMode.FULL_SET: STATE_ALARM_ARMED_AWAY, + } + return mode_to_state.get(area.mode) -@asyncio.coroutine -def async_setup_platform(hass, config, async_add_entities, - discovery_info=None): +async 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 - api = hass.data[DATA_API] - devices = [SpcAlarm(api, area) - for area in discovery_info[ATTR_DISCOVER_AREAS]] - - async_add_entities(devices) + async_add_entities([SpcAlarm(area=area, api=hass.data[DATA_API]) + for area in discovery_info[ATTR_DISCOVER_AREAS]]) class SpcAlarm(alarm.AlarmControlPanel): """Representation of the SPC alarm panel.""" - def __init__(self, api, area): + def __init__(self, area, api): """Initialize the SPC alarm panel.""" - self._area_id = area['id'] - self._name = area['name'] - self._state = _get_alarm_state(area['mode']) - if self._state == STATE_ALARM_DISARMED: - self._changed_by = area.get('last_unset_user_name', 'unknown') - else: - self._changed_by = area.get('last_set_user_name', 'unknown') + self._area = area self._api = api - @asyncio.coroutine - def async_added_to_hass(self): + async def async_added_to_hass(self): """Call for adding new entities.""" - self.hass.data[DATA_REGISTRY].register_alarm_device( - self._area_id, self) + async_dispatcher_connect(self.hass, + SIGNAL_UPDATE_ALARM.format(self._area.id), + self._update_callback) - @asyncio.coroutine - def async_update_from_spc(self, state, extra): - """Update the alarm panel with a new state.""" - self._state = state - self._changed_by = extra.get('changed_by', 'unknown') - self.async_schedule_update_ha_state() + @callback + def _update_callback(self): + """Call update method.""" + self.async_schedule_update_ha_state(True) @property def should_poll(self): @@ -78,32 +72,34 @@ class SpcAlarm(alarm.AlarmControlPanel): @property def name(self): """Return the name of the device.""" - return self._name + return self._area.name @property def changed_by(self): """Return the user the last change was triggered by.""" - return self._changed_by + return self._area.last_changed_by @property def state(self): """Return the state of the device.""" - return self._state + return _get_alarm_state(self._area) - @asyncio.coroutine - def async_alarm_disarm(self, code=None): + async def async_alarm_disarm(self, code=None): """Send disarm command.""" - yield from self._api.send_area_command( - self._area_id, SpcWebGateway.AREA_COMMAND_UNSET) + from pyspcwebgw.const import AreaMode + self._api.change_mode(area=self._area, new_mode=AreaMode.UNSET) - @asyncio.coroutine - def async_alarm_arm_home(self, code=None): + async 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) + from pyspcwebgw.const import AreaMode + self._api.change_mode(area=self._area, new_mode=AreaMode.PART_SET_A) - @asyncio.coroutine - def async_alarm_arm_away(self, code=None): + async def async_alarm_arm_night(self, code=None): + """Send arm home command.""" + from pyspcwebgw.const import AreaMode + self._api.change_mode(area=self._area, new_mode=AreaMode.PART_SET_B) + + async 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) + from pyspcwebgw.const import AreaMode + self._api.change_mode(area=self._area, new_mode=AreaMode.FULL_SET) diff --git a/homeassistant/components/binary_sensor/spc.py b/homeassistant/components/binary_sensor/spc.py index 9afd4fe4015..c1be72db374 100644 --- a/homeassistant/components/binary_sensor/spc.py +++ b/homeassistant/components/binary_sensor/spc.py @@ -4,87 +4,66 @@ 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 asyncio import logging from homeassistant.components.binary_sensor import BinarySensorDevice -from homeassistant.components.spc import ATTR_DISCOVER_DEVICES, DATA_REGISTRY -from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.core import callback +from homeassistant.components.spc import ( + ATTR_DISCOVER_DEVICES, SIGNAL_UPDATE_SENSOR) _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(zone_type): + from pyspcwebgw.const import ZoneType + return { + ZoneType.ALARM: 'motion', + ZoneType.ENTRY_EXIT: 'opening', + ZoneType.FIRE: 'smoke', + }.get(zone_type) -def _get_device_class(spc_type): - """Get the device class.""" - return SPC_TYPE_TO_DEVICE_CLASS.get(spc_type, None) - - -def _get_sensor_state(spc_input): - """Get the sensor state.""" - return SPC_INPUT_TO_SENSOR_STATE.get(spc_input, STATE_UNAVAILABLE) - - -def _create_sensor(hass, zone): - """Create a SPC sensor.""" - 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): +async def async_setup_platform(hass, config, async_add_entities, + discovery_info=None): """Set up the SPC binary sensor.""" 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'])) + async_add_entities(SpcBinarySensor(zone) + for zone in discovery_info[ATTR_DISCOVER_DEVICES] + if _get_device_class(zone.type)) class SpcBinarySensor(BinarySensorDevice): """Representation of a sensor based on a SPC zone.""" - def __init__(self, zone_id, name, state, device_class, spc_registry): + def __init__(self, zone): """Initialize the sensor device.""" - self._zone_id = zone_id - self._name = name - self._state = state - self._device_class = device_class + self._zone = zone - spc_registry.register_sensor_device(zone_id, self) + async def async_added_to_hass(self): + """Call for adding new entities.""" + async_dispatcher_connect(self.hass, + SIGNAL_UPDATE_SENSOR.format(self._zone.id), + self._update_callback) - @asyncio.coroutine - def async_update_from_spc(self, state, extra): - """Update the state of the device.""" - self._state = state - yield from self.async_update_ha_state() + @callback + def _update_callback(self): + """Call update method.""" + self.async_schedule_update_ha_state(True) @property def name(self): """Return the name of the device.""" - return self._name + return self._zone.name @property def is_on(self): """Whether the device is switched on.""" - return self._state == STATE_ON + from pyspcwebgw.const import ZoneInput + return self._zone.input == ZoneInput.OPEN @property def hidden(self) -> bool: @@ -100,4 +79,4 @@ class SpcBinarySensor(BinarySensorDevice): @property def device_class(self): """Return the device class.""" - return self._device_class + return _get_device_class(self._zone.type) diff --git a/homeassistant/components/spc.py b/homeassistant/components/spc.py index 5b357efcabd..b00a4aeed2c 100644 --- a/homeassistant/components/spc.py +++ b/homeassistant/components/spc.py @@ -4,23 +4,15 @@ 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 asyncio -import json import logging -from urllib.parse import urljoin -import aiohttp -import async_timeout import voluptuous as vol -from homeassistant.const import ( - STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED, - STATE_ALARM_TRIGGERED, STATE_OFF, STATE_ON, STATE_UNAVAILABLE, - STATE_UNKNOWN) -from homeassistant.helpers import discovery +from homeassistant.helpers import discovery, aiohttp_client +from homeassistant.helpers.dispatcher import async_dispatcher_send import homeassistant.helpers.config_validation as cv -REQUIREMENTS = ['websockets==6.0'] +REQUIREMENTS = ['pyspcwebgw==0.4.0'] _LOGGER = logging.getLogger(__name__) @@ -30,9 +22,11 @@ ATTR_DISCOVER_AREAS = 'areas' CONF_WS_URL = 'ws_url' CONF_API_URL = 'api_url' -DATA_REGISTRY = 'spc_registry' -DATA_API = 'spc_api' DOMAIN = 'spc' +DATA_API = 'spc_api' + +SIGNAL_UPDATE_ALARM = 'spc_update_alarm_{}' +SIGNAL_UPDATE_SENSOR = 'spc_update_sensor_{}' CONFIG_SCHEMA = vol.Schema({ DOMAIN: vol.Schema({ @@ -42,244 +36,45 @@ CONFIG_SCHEMA = vol.Schema({ }, extra=vol.ALLOW_EXTRA) -@asyncio.coroutine -def async_setup(hass, config): - """Set up the SPC platform.""" - hass.data[DATA_REGISTRY] = SpcRegistry() +async def async_setup(hass, config): + """Set up the SPC component.""" + from pyspcwebgw import SpcWebGateway - api = SpcWebGateway(hass, - config[DOMAIN].get(CONF_API_URL), - config[DOMAIN].get(CONF_WS_URL)) + async def async_upate_callback(spc_object): + from pyspcwebgw.area import Area + from pyspcwebgw.zone import Zone - hass.data[DATA_API] = api + if isinstance(spc_object, Area): + async_dispatcher_send(hass, + SIGNAL_UPDATE_ALARM.format(spc_object.id)) + elif isinstance(spc_object, Zone): + async_dispatcher_send(hass, + SIGNAL_UPDATE_SENSOR.format(spc_object.id)) + + session = aiohttp_client.async_get_clientsession(hass) + + spc = SpcWebGateway(loop=hass.loop, session=session, + api_url=config[DOMAIN].get(CONF_API_URL), + ws_url=config[DOMAIN].get(CONF_WS_URL), + async_callback=async_upate_callback) + + hass.data[DATA_API] = spc + + if not await spc.async_load_parameters(): + _LOGGER.error('Failed to load area/zone information from SPC.') + return False # add sensor devices for each zone (typically motion/fire/door sensors) - zones = yield from api.get_zones() - if zones: - hass.async_create_task(discovery.async_load_platform( - hass, 'binary_sensor', DOMAIN, - {ATTR_DISCOVER_DEVICES: zones}, config)) + hass.async_create_task(discovery.async_load_platform( + hass, 'binary_sensor', DOMAIN, + {ATTR_DISCOVER_DEVICES: spc.zones.values()}, config)) # create a separate alarm panel for each area - areas = yield from api.get_areas() - if areas: - hass.async_create_task(discovery.async_load_platform( - hass, 'alarm_control_panel', DOMAIN, - {ATTR_DISCOVER_AREAS: areas}, config)) + hass.async_create_task(discovery.async_load_platform( + hass, 'alarm_control_panel', DOMAIN, + {ATTR_DISCOVER_AREAS: spc.areas.values()}, config)) # start listening for incoming events over websocket - api.start_listener(_async_process_message, hass.data[DATA_REGISTRY]) + spc.start() 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 - - extra = {} - - if sia_code in ('BA', 'CG', 'NL', 'OG'): - # change in area status, notify alarm panel device - device = spc_registry.get_alarm_device(spc_id) - data = sia_message['description'].split('¦') - if len(data) == 3: - extra['changed_by'] = data[1] - 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, - '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, extra) - - -class SpcRegistry: - """Maintain 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: # noqa: E722 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.""" - asyncio.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): - """Get the data from the 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): - """Call web gateway for data.""" - response = None - session = None - url = self._build_url(resource) - try: - _LOGGER.debug("Attempting to retrieve SPC data from %s", url) - session = \ - self._hass.helpers.aiohttp_client.async_get_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): - """Read from websocket.""" - 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): - """Listen on websocket.""" - 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 d901a787493..ba353af0d48 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1069,6 +1069,9 @@ pysnmp==4.4.5 # homeassistant.components.sonos pysonos==0.0.2 +# homeassistant.components.spc +pyspcwebgw==0.4.0 + # homeassistant.components.notify.stride pystride==0.1.7 @@ -1502,7 +1505,6 @@ waterfurnace==0.7.0 # homeassistant.components.media_player.gpmdp websocket-client==0.37.0 -# homeassistant.components.spc # homeassistant.components.media_player.webostv websockets==6.0 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index dc263fd4e9c..171650b867d 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -170,6 +170,9 @@ pyqwikswitch==0.8 # homeassistant.components.sonos pysonos==0.0.2 +# homeassistant.components.spc +pyspcwebgw==0.4.0 + # homeassistant.components.sensor.darksky # homeassistant.components.weather.darksky python-forecastio==1.4.0 diff --git a/script/gen_requirements_all.py b/script/gen_requirements_all.py index 90036c5bf33..7493e523273 100755 --- a/script/gen_requirements_all.py +++ b/script/gen_requirements_all.py @@ -83,6 +83,7 @@ TEST_REQUIREMENTS = ( 'pysonos', 'pyqwikswitch', 'PyRMVtransport', + 'pyspcwebgw', 'python-forecastio', 'python-nest', 'pytradfri\[async\]', diff --git a/tests/components/alarm_control_panel/test_spc.py b/tests/components/alarm_control_panel/test_spc.py index a0793943c2f..b1078e1b14f 100644 --- a/tests/components/alarm_control_panel/test_spc.py +++ b/tests/components/alarm_control_panel/test_spc.py @@ -1,27 +1,11 @@ """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) +from homeassistant.components.spc import (DATA_API) -@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(force=True)) - - -@asyncio.coroutine -def test_setup_platform(hass): +async def test_setup_platform(hass): """Test adding areas as separate alarm control panel devices.""" added_entities = [] @@ -29,7 +13,7 @@ def test_setup_platform(hass): nonlocal added_entities added_entities = list(entities) - areas = {'areas': [{ + area_defs = [{ 'id': '1', 'name': 'House', 'mode': '3', @@ -50,12 +34,18 @@ def test_setup_platform(hass): '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) + from pyspcwebgw import Area + + areas = [Area(gateway=None, spc_area=a) for a in area_defs] + + hass.data[DATA_API] = None + + await spc.async_setup_platform(hass=hass, + config={}, + async_add_entities=add_entities, + discovery_info={'areas': areas}) assert len(added_entities) == 2 diff --git a/tests/components/binary_sensor/test_spc.py b/tests/components/binary_sensor/test_spc.py index 966f73682e8..ec0886aeed8 100644 --- a/tests/components/binary_sensor/test_spc.py +++ b/tests/components/binary_sensor/test_spc.py @@ -1,28 +1,12 @@ """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(force=True)) - - -@asyncio.coroutine -def test_setup_platform(hass): +async def test_setup_platform(hass): """Test autodiscovery of supported device types.""" added_entities = [] - zones = {'devices': [{ + zone_defs = [{ 'id': '1', 'type': '3', 'zone_name': 'Kitchen smoke', @@ -46,16 +30,20 @@ def test_setup_platform(hass): '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) + from pyspcwebgw import Zone + + zones = [Zone(area=None, spc_zone=z) for z in zone_defs] + + await spc.async_setup_platform(hass=hass, + config={}, + async_add_entities=add_entities, + discovery_info={'devices': zones}) assert len(added_entities) == 3 assert added_entities[0].device_class == 'smoke' diff --git a/tests/components/test_spc.py b/tests/components/test_spc.py index 7837abd8007..d4bedda4e96 100644 --- a/tests/components/test_spc.py +++ b/tests/components/test_spc.py @@ -1,167 +1,74 @@ """Tests for Vanderbilt SPC component.""" -import asyncio +from unittest.mock import patch, PropertyMock, Mock -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_ON, STATE_OFF, STATE_ALARM_ARMED_AWAY, - STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED) +from homeassistant.components.spc import DATA_API +from homeassistant.const import (STATE_ALARM_ARMED_AWAY, STATE_ALARM_DISARMED) + +from tests.common import mock_coro -@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 -@pytest.mark.parametrize("sia_code,state", [ - ('NL', STATE_ALARM_ARMED_HOME), - ('CG', STATE_ALARM_ARMED_AWAY), - ('OG', STATE_ALARM_DISARMED) -]) -def test_update_alarm_device(hass, aioclient_mock, monkeypatch, - sia_code, state): - """Test that alarm panel state changes on incoming websocket data.""" - monkeypatch.setattr("homeassistant.components.spc.SpcWebGateway." - "start_listener", lambda x, *args: None) +async def test_valid_device_config(hass, monkeypatch): + """Test valid device config.""" 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() + with patch('pyspcwebgw.SpcWebGateway.async_load_parameters', + return_value=mock_coro(True)): + assert await async_setup_component(hass, 'spc', config) is True + + +async def test_invalid_device_config(hass, monkeypatch): + """Test valid device config.""" + config = { + 'spc': { + 'api_url': 'http://localhost/' + } + } + + with patch('pyspcwebgw.SpcWebGateway.async_load_parameters', + return_value=mock_coro(True)): + assert await async_setup_component(hass, 'spc', config) is False + + +async def test_update_alarm_device(hass): + """Test that alarm panel state changes on incoming websocket data.""" + import pyspcwebgw + from pyspcwebgw.const import AreaMode + + config = { + 'spc': { + 'api_url': 'http://localhost/', + 'ws_url': 'ws://localhost/' + } + } + + area_mock = Mock(spec=pyspcwebgw.area.Area, id='1', + mode=AreaMode.FULL_SET, last_changed_by='Sven') + area_mock.name = 'House' + area_mock.verified_alarm = False + + with patch('pyspcwebgw.SpcWebGateway.areas', + new_callable=PropertyMock) as mock_areas: + mock_areas.return_value = {'1': area_mock} + with patch('pyspcwebgw.SpcWebGateway.async_load_parameters', + return_value=mock_coro(True)): + assert await async_setup_component(hass, 'spc', config) is True + + await hass.async_block_till_done() entity_id = 'alarm_control_panel.house' + assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY + assert hass.states.get(entity_id).attributes['changed_by'] == 'Sven' + + area_mock.mode = AreaMode.UNSET + area_mock.last_changed_by = 'Anna' + await hass.data[DATA_API]._async_callback(area_mock) + await hass.async_block_till_done() + assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED - - msg = {"sia_code": sia_code, "sia_address": "1", - "description": "House¦Sam¦1"} - yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY]) - yield from hass.async_block_till_done() - - state_obj = hass.states.get(entity_id) - assert state_obj.state == state - assert state_obj.attributes['changed_by'] == 'Sam' - - -@asyncio.coroutine -@pytest.mark.parametrize("sia_code,state", [ - ('ZO', STATE_ON), - ('ZC', STATE_OFF) -]) -def test_update_sensor_device(hass, aioclient_mock, monkeypatch, - sia_code, state): - """ - Test that sensors change state on incoming websocket data. - - Note that we don't test for the ZD (disconnected) and ZX (problem/short) - codes since the binary sensor component is hardcoded to only - let on/off states through. - """ - 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 == STATE_OFF - - msg = {"sia_code": sia_code, "sia_address": "3", - "description": "Hallway PIR"} - yield from spc._async_process_message(msg, hass.data[spc.DATA_REGISTRY]) - yield from hass.async_block_till_done() - assert hass.states.get('binary_sensor.hallway_pir').state == state - - -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 + assert hass.states.get(entity_id).attributes['changed_by'] == 'Anna'