hass-core/tests/components/test_rflink.py
Johan Bloemberg bbda2a72f4 Rflink 433Mhz gateway platform and components (#4547)
* Initial sketches of rflink component.

* Add requirement.

* Properly load configuration.

* Bump rflink for graceful parse errors and protocol callback.

* Cleanup, documentation and linting.

* More documentation, first sensor implementation (temp & hum).

* Add brightness/dim support for newkaku protocol.

* Use separate class for dimmables.

* Make sure non-dimmable newkaku devices are turned on.

* Move some code around, add switches. Support loading from config.

* Fix bug in ignoring devices.

* Fix initial state assumption.

* Improve reliability on invalid conditions.

* Allow configuration of group for new devices.

* Sensor icons.

* Fix parsing negative numbers.

* Correct icon.

* Allow sending commands serial.

* Pluralize.

* Allow adding sensors from config.

* Fix ignoring devices and bugs in previous commit.

* Share know devices so devices from configuration don't get added as lights.

* Lookup unit from value_key.

* Remove debug.

* Start implementing event protocol in place of packet protocol.

- Added first test suite for sensors.
- This currently breaks light and switch.

* Refactor switch component to fit new rflink changes. Add test suite.

* Fix style.

* Refactor and test lights. Bring coverage to 100%.

* Use non-broken and production tested rflink module.

* Update requirements.

* Bump for logging.

* Improve readability.

* Do not use global variable but keep known device state in intended place.

* Improve docs.

* Make icon support generic.

* Disable overriding icons in config, as it belongs in customization. Only keep custom icon for entities that are able to detect a icon based on the thing they represent (sensors in this case).

* Implement configuration schema, overall refactor of magic values.

* Fix bug in config/test wait_for_ack.

* Small refactors.

* Move command logic into separate class.

* Convert command sending logic to class based pattern instead of using the event bus.

* Start not using bus for rflink event propagation to platforms.

* Do not use event bus for all entity types.

* Fire an event on the bus for every switch incoming rflink command.

* Resolve lint errors, remove some old code.

* Known devices no longer need to be registered separately.

* Log bus events.

* Event callback is a..... callback.

* Use full entity id for events.

* Move event sending to entity.

* Log incoming events.

* Make firing events optional inline with rfxtrx.

* Add foundation for signal repetition.

* Add signal repetition config and tests.

* Make plain switchable type explicitly configurable.

* Enable default entity settings for automatically added entities as well.

* Prevent default configuration leaking accross entities.

* Make sure device defaults don't get overwritten by defaults further down.

* Don't let fast state switching and repetitions turn your house into a disco.

* Make repetitions more responsive.

* Disable on/off fallback on dimmables as it currently doesn't play nice with repetitions.

* Use rflink that allows send_command_ack to be safely cancelled.

* Reduce duplication and make repeat work for non-ack.

* Implement reconnection logic.

* Improve reconnection logic.

* Also cancel repetitions when entity state is changed due to external command.

* Update requirements.

* Fix linting.

* Fix spelling.

* Don't lie.

* Fix lint.

* Support for automatically creating protocol translation (fixes spaces in device names).

* Returned support for dimmable and on/off entity.

* Duplicate code to fix linting issues with inheritance.

* Allow overriding unit of measurement from config.
2017-01-31 08:11:52 -08:00

178 lines
5.2 KiB
Python

"""Common functions for Rflink component tests and generic platform tests."""
import asyncio
from unittest.mock import Mock
from homeassistant.bootstrap import async_setup_component
from homeassistant.components.rflink import CONF_RECONNECT_INTERVAL
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF
from tests.common import assert_setup_component
@asyncio.coroutine
def mock_rflink(hass, config, domain, monkeypatch, failures=None):
"""Create mock Rflink asyncio protocol, test component setup."""
transport, protocol = (Mock(), Mock())
@asyncio.coroutine
def send_command_ack(*command):
return True
protocol.send_command_ack = Mock(wraps=send_command_ack)
@asyncio.coroutine
def send_command(*command):
return True
protocol.send_command = Mock(wraps=send_command)
@asyncio.coroutine
def create_rflink_connection(*args, **kwargs):
"""Return mocked transport and protocol."""
# failures can be a list of booleans indicating in which sequence
# creating a connection should success or fail
if failures:
fail = failures.pop()
else:
fail = False
if fail:
raise ConnectionRefusedError
else:
return transport, protocol
mock_create = Mock(wraps=create_rflink_connection)
monkeypatch.setattr(
'rflink.protocol.create_rflink_connection',
mock_create)
# verify instanstiation of component with given config
with assert_setup_component(1, domain):
yield from async_setup_component(hass, domain, config)
# hook into mock config for injecting events
event_callback = mock_create.call_args_list[0][1]['event_callback']
assert event_callback
disconnect_callback = mock_create.call_args_list[
0][1]['disconnect_callback']
return event_callback, mock_create, protocol, disconnect_callback
@asyncio.coroutine
def test_version_banner(hass, monkeypatch):
"""Test sending unknown commands doesn't cause issues."""
# use sensor domain during testing main platform
domain = 'sensor'
config = {
'rflink': {'port': '/dev/ttyABC0', },
domain: {
'platform': 'rflink',
'devices': {
'test': {'name': 'test', 'sensor_type': 'temperature', },
},
},
}
# setup mocking rflink module
event_callback, _, _, _ = yield from mock_rflink(
hass, config, domain, monkeypatch)
event_callback({
'hardware': 'Nodo RadioFrequencyLink',
'firmware': 'RFLink Gateway',
'version': '1.1',
'revision': '45',
})
@asyncio.coroutine
def test_send_no_wait(hass, monkeypatch):
"""Test command sending without ack."""
domain = 'switch'
config = {
'rflink': {
'port': '/dev/ttyABC0',
'wait_for_ack': False,
},
domain: {
'platform': 'rflink',
'devices': {
'protocol_0_0': {
'name': 'test',
'aliasses': ['test_alias_0_0'],
},
},
},
}
# setup mocking rflink module
_, _, protocol, _ = yield from mock_rflink(
hass, config, domain, monkeypatch)
hass.async_add_job(
hass.services.async_call(domain, SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: 'switch.test'}))
yield from hass.async_block_till_done()
assert protocol.send_command.call_args_list[0][0][0] == 'protocol_0_0'
assert protocol.send_command.call_args_list[0][0][1] == 'off'
@asyncio.coroutine
def test_reconnecting_after_disconnect(hass, monkeypatch):
"""An unexpected disconnect should cause a reconnect."""
domain = 'sensor'
config = {
'rflink': {
'port': '/dev/ttyABC0',
CONF_RECONNECT_INTERVAL: 0,
},
domain: {
'platform': 'rflink',
},
}
# setup mocking rflink module
_, mock_create, _, disconnect_callback = yield from mock_rflink(
hass, config, domain, monkeypatch)
assert disconnect_callback, 'disconnect callback not passed to rflink'
# rflink initiated disconnect
disconnect_callback(None)
yield from hass.async_block_till_done()
# we expect 2 call, the initial and reconnect
assert mock_create.call_count == 2
@asyncio.coroutine
def test_reconnecting_after_failure(hass, monkeypatch):
"""A failure to reconnect should be retried."""
domain = 'sensor'
config = {
'rflink': {
'port': '/dev/ttyABC0',
CONF_RECONNECT_INTERVAL: 0,
},
domain: {
'platform': 'rflink',
},
}
# success first time but fail second
failures = [False, True, False]
# setup mocking rflink module
_, mock_create, _, disconnect_callback = yield from mock_rflink(
hass, config, domain, monkeypatch, failures=failures)
# rflink initiated disconnect
disconnect_callback(None)
# wait for reconnects to have happened
yield from hass.async_block_till_done()
yield from hass.async_block_till_done()
# we expect 3 calls, the initial and 2 reconnects
assert mock_create.call_count == 3