Qwikswitch binary sensors (#14008)

This commit is contained in:
Johann Kellerman
2018-04-24 23:19:15 -04:00
committed by Paulus Schoutsen
parent 8cb1e17ad8
commit 44be80145b
6 changed files with 160 additions and 40 deletions
@@ -13,17 +13,19 @@ _LOGGER = logging.getLogger(__name__)
class AiohttpClientMockResponseList(list):
"""List that fires an event on empty pop, for aiohttp Mocker."""
"""Return multiple values for aiohttp Mocker.
aoihttp mocker uses decode to fetch the next value.
"""
def decode(self, _):
"""Return next item from list."""
try:
res = list.pop(self)
res = list.pop(self, 0)
_LOGGER.debug("MockResponseList popped %s: %s", res, self)
return res
except IndexError:
_LOGGER.debug("MockResponseList empty")
return ""
raise AssertionError("MockResponseList empty")
async def wait_till_empty(self, hass):
"""Wait until empty."""
@@ -52,8 +54,8 @@ def aioclient_mock():
yield mock_session
async def test_sensor_device(hass, aioclient_mock):
"""Test a sensor device."""
async def test_binary_sensor_device(hass, aioclient_mock):
"""Test a binary sensor device."""
config = {
'qwikswitch': {
'sensors': {
@@ -67,21 +69,49 @@ async def test_sensor_device(hass, aioclient_mock):
await async_setup_component(hass, QWIKSWITCH, config)
await hass.async_block_till_done()
state_obj = hass.states.get('sensor.s1')
assert state_obj
assert state_obj.state == 'None'
state_obj = hass.states.get('binary_sensor.s1')
assert state_obj.state == 'off'
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
LISTEN.append( # Close
"""{"id":"@a00001","cmd":"","data":"4e0e1601","rssi":"61%"}""")
LISTEN.append('{"id":"@a00001","cmd":"","data":"4e0e1601","rssi":"61%"}')
LISTEN.append('') # Will cause a sleep
await hass.async_block_till_done()
state_obj = hass.states.get('sensor.s1')
assert state_obj.state == 'True'
state_obj = hass.states.get('binary_sensor.s1')
assert state_obj.state == 'on'
# Causes a 30second delay: can be uncommented when upstream library
# allows cancellation of asyncio.sleep(30) on failed packet ("")
# LISTEN.append( # Open
# """{"id":"@a00001","cmd":"","data":"4e0e1701","rssi":"61%"}""")
# await LISTEN.wait_till_empty(hass)
# state_obj = hass.states.get('sensor.s1')
# assert state_obj.state == 'False'
LISTEN.append('{"id":"@a00001","cmd":"","data":"4e0e1701","rssi":"61%"}')
hass.data[QWIKSWITCH]._sleep_task.cancel()
await LISTEN.wait_till_empty(hass)
state_obj = hass.states.get('binary_sensor.s1')
assert state_obj.state == 'off'
async def test_sensor_device(hass, aioclient_mock):
"""Test a sensor device."""
config = {
'qwikswitch': {
'sensors': {
'name': 'ss1',
'id': '@a00001',
'channel': 1,
'type': 'qwikcord',
}
}
}
await async_setup_component(hass, QWIKSWITCH, config)
await hass.async_block_till_done()
state_obj = hass.states.get('sensor.ss1')
assert state_obj.state == 'None'
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
LISTEN.append(
'{"id":"@a00001","name":"ss1","type":"rel",'
'"val":"4733800001a00000"}')
LISTEN.append('') # Will cause a sleep
await LISTEN.wait_till_empty(hass) # await hass.async_block_till_done()
state_obj = hass.states.get('sensor.ss1')
assert state_obj.state == 'None'