Use pyspcwebgw for SPC component (#16214)
* Use pyspcwebgw library. * Support alarm triggering. * Update requirements. * Add pyspcwebgw to test reqs. * Also update script. * Use dispatcher. * Address review feedback.
This commit is contained in:
Martin Berg
Martin Hjelmare
parent
4fd2f773ad
commit
a5cb4e6c2b
9 changed files with 210 additions and 549 deletions
|
|
@ -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'
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue