jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup tests for lcn events and device triggers (#61719)

Browse Source 
* Return PchkConnectionManager instance from init_integration

* Removed ip and port from LCN host model identifer

* Fix syntax error

* Convert init_integration to a fixture

* Rename device model for host

* Instantiate MockPchkConnectionManager with arguments from tests

* Invert logic for testing devices
This commit is contained in:
Andre Lengwenus 2021-12-19 10:38:33 +01:00 committed by GitHub
parent 615872a5d1
commit 38cb477e7b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 78 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/lcn/device_trigger.py
View File

@ -58,8 +58,11 @@ async def async_get_triggers(
"""List device triggers for LCN devices."""
device_registry = dr.async_get(hass)
device = device_registry.async_get(device_id)
if device is None:
return []
if device.model.startswith(("LCN host", "LCN group", "LCN resource")): # type: ignore[union-attr]
identifier = next(iter(device.identifiers))
if (identifier[1].count("-") != 1) or device.model.startswith("LCN group"): # type: ignore[union-attr]
return []
base_trigger = {

2
homeassistant/components/lcn/helpers.py
View File

@ -256,7 +256,7 @@ def register_lcn_host_device(hass: HomeAssistant, config_entry: ConfigEntry) ->
identifiers={(DOMAIN, config_entry.entry_id)},
manufacturer="Issendorff",
name=config_entry.title,
model=f"LCN host ({config_entry.data[CONF_IP_ADDRESS]}:{config_entry.data[CONF_PORT]})",
model="LCN-PCHK",
)

28
tests/components/lcn/conftest.py
View File

@ -41,13 +41,6 @@ class MockGroupConnection(GroupConnection):
class MockPchkConnectionManager(PchkConnectionManager):
"""Fake connection handler."""
return_value = None
def __init__(self, *args, **kwargs):
"""Initialize MockPchkCOnnectionManager."""
super().__init__(*args, **kwargs)
self.__class__.return_value = self
async def async_connect(self, timeout=30):
"""Mock establishing a connection to PCHK."""
self.authentication_completed_future.set_result(True)
@ -59,9 +52,9 @@ class MockPchkConnectionManager(PchkConnectionManager):
@patch.object(pypck.connection, "ModuleConnection", MockModuleConnection)
@patch.object(pypck.connection, "GroupConnection", MockGroupConnection)
def get_address_conn(self, addr):
def get_address_conn(self, addr, request_serials=False):
"""Get LCN address connection."""
return super().get_address_conn(addr, request_serials=False)
return super().get_address_conn(addr, request_serials)
send_command = AsyncMock()
@ -102,11 +95,24 @@ def create_config_entry_myhome():
return create_config_entry("myhome")
@pytest.fixture(name="lcn_connection")
async def init_integration(hass, entry):
"""Set up the LCN integration in Home Assistant."""
lcn_connection = None
def lcn_connection_factory(*args, **kwargs):
nonlocal lcn_connection
lcn_connection = MockPchkConnectionManager(*args, **kwargs)
return lcn_connection
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch(
"pypck.connection.PchkConnectionManager",
side_effect=lcn_connection_factory,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
yield lcn_connection
async def setup_component(hass):

68
tests/components/lcn/test_device_trigger.py
View File

@ -1,6 +1,4 @@
"""Tests for LCN device triggers."""
from unittest.mock import patch
from pypck.inputs import ModSendKeysHost, ModStatusAccessControl
from pypck.lcn_addr import LcnAddr
from pypck.lcn_defs import AccessControlPeriphery, KeyAction, SendKeyCommand
@ -13,15 +11,13 @@ from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.setup import async_setup_component
from .conftest import MockPchkConnectionManager, get_device, init_integration
from .conftest import get_device
from tests.common import assert_lists_same, async_get_device_automations
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_get_triggers_module_device(hass, entry):
async def test_get_triggers_module_device(hass, entry, lcn_connection):
"""Test we get the expected triggers from a LCN module device."""
await init_integration(hass, entry)
device = get_device(hass, entry, (0, 7, False))
expected_triggers = [
@ -55,25 +51,25 @@ async def test_get_triggers_module_device(hass, entry):
assert_lists_same(triggers, expected_triggers)
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_get_triggers_non_module_device(hass, entry):
async def test_get_triggers_non_module_device(hass, entry, lcn_connection):
"""Test we get the expected triggers from a LCN non-module device."""
not_included_types = ("transmitter", "transponder", "fingerprint", "send_keys")
await init_integration(hass, entry)
device_registry = dr.async_get(hass)
for device_id in device_registry.devices:
device = device_registry.async_get(device_id)
if device.model.startswith(("LCN host", "LCN group", "LCN resource")):
triggers = await async_get_device_automations(hass, "trigger", device_id)
for trigger in triggers:
assert trigger[CONF_TYPE] not in not_included_types
host_device = device_registry.async_get_device({(DOMAIN, entry.entry_id)})
group_device = get_device(hass, entry, (0, 5, True))
resource_device = device_registry.async_get_device(
{(DOMAIN, f"{entry.entry_id}-m000007-output1")}
)
for device in (host_device, group_device, resource_device):
triggers = await async_get_device_automations(hass, "trigger", device.id)
for trigger in triggers:
assert trigger[CONF_TYPE] not in not_included_types
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_if_fires_on_transponder_event(hass, calls, entry):
async def test_if_fires_on_transponder_event(hass, calls, entry, lcn_connection):
"""Test for transponder event triggers firing."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -107,7 +103,6 @@ async def test_if_fires_on_transponder_event(hass, calls, entry):
code="aabbcc",
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -118,10 +113,8 @@ async def test_if_fires_on_transponder_event(hass, calls, entry):
}
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_if_fires_on_fingerprint_event(hass, calls, entry):
async def test_if_fires_on_fingerprint_event(hass, calls, entry, lcn_connection):
"""Test for fingerprint event triggers firing."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -155,7 +148,6 @@ async def test_if_fires_on_fingerprint_event(hass, calls, entry):
code="aabbcc",
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -166,10 +158,8 @@ async def test_if_fires_on_fingerprint_event(hass, calls, entry):
}
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_if_fires_on_transmitter_event(hass, calls, entry):
async def test_if_fires_on_transmitter_event(hass, calls, entry, lcn_connection):
"""Test for transmitter event triggers firing."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -209,7 +199,6 @@ async def test_if_fires_on_transmitter_event(hass, calls, entry):
action=KeyAction.HIT,
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -223,10 +212,8 @@ async def test_if_fires_on_transmitter_event(hass, calls, entry):
}
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_if_fires_on_send_keys_event(hass, calls, entry):
async def test_if_fires_on_send_keys_event(hass, calls, entry, lcn_connection):
"""Test for send_keys event triggers firing."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -261,7 +248,6 @@ async def test_if_fires_on_send_keys_event(hass, calls, entry):
keys=[True, False, False, False, False, False, False, False],
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -273,10 +259,8 @@ async def test_if_fires_on_send_keys_event(hass, calls, entry):
}
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_get_transponder_trigger_capabilities(hass, entry):
async def test_get_transponder_trigger_capabilities(hass, entry, lcn_connection):
"""Test we get the expected capabilities from a transponder device trigger."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -296,10 +280,8 @@ async def test_get_transponder_trigger_capabilities(hass, entry):
) == [{"name": "code", "optional": True, "type": "string", "lower": True}]
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_get_fingerprint_trigger_capabilities(hass, entry):
async def test_get_fingerprint_trigger_capabilities(hass, entry, lcn_connection):
"""Test we get the expected capabilities from a fingerprint device trigger."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -319,10 +301,8 @@ async def test_get_fingerprint_trigger_capabilities(hass, entry):
) == [{"name": "code", "optional": True, "type": "string", "lower": True}]
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_get_transmitter_trigger_capabilities(hass, entry):
async def test_get_transmitter_trigger_capabilities(hass, entry, lcn_connection):
"""Test we get the expected capabilities from a transmitter device trigger."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -352,10 +332,8 @@ async def test_get_transmitter_trigger_capabilities(hass, entry):
]
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_get_send_keys_trigger_capabilities(hass, entry):
async def test_get_send_keys_trigger_capabilities(hass, entry, lcn_connection):
"""Test we get the expected capabilities from a send_keys device trigger."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)
@ -390,10 +368,8 @@ async def test_get_send_keys_trigger_capabilities(hass, entry):
]
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_unknown_trigger_capabilities(hass, entry):
async def test_unknown_trigger_capabilities(hass, entry, lcn_connection):
"""Test we get empty capabilities if trigger is unknown."""
await init_integration(hass, entry)
address = (0, 7, False)
device = get_device(hass, entry, address)

43
tests/components/lcn/test_events.py
View File

@ -1,20 +1,13 @@
"""Tests for LCN events."""
from unittest.mock import patch
from pypck.inputs import Input, ModSendKeysHost, ModStatusAccessControl
from pypck.lcn_addr import LcnAddr
from pypck.lcn_defs import AccessControlPeriphery, KeyAction, SendKeyCommand
from .conftest import MockPchkConnectionManager, init_integration
from tests.common import async_capture_events
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_fire_transponder_event(hass, entry):
async def test_fire_transponder_event(hass, lcn_connection):
"""Test the transponder event is fired."""
await init_integration(hass, entry)
events = async_capture_events(hass, "lcn_transponder")
inp = ModStatusAccessControl(
@ -23,7 +16,6 @@ async def test_fire_transponder_event(hass, entry):
code="aabbcc",
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -32,11 +24,8 @@ async def test_fire_transponder_event(hass, entry):
assert events[0].data["code"] == "aabbcc"
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_fire_fingerprint_event(hass, entry):
async def test_fire_fingerprint_event(hass, lcn_connection):
"""Test the fingerprint event is fired."""
await init_integration(hass, entry)
events = async_capture_events(hass, "lcn_fingerprint")
inp = ModStatusAccessControl(
@ -45,7 +34,6 @@ async def test_fire_fingerprint_event(hass, entry):
code="aabbcc",
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -54,11 +42,8 @@ async def test_fire_fingerprint_event(hass, entry):
assert events[0].data["code"] == "aabbcc"
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_fire_transmitter_event(hass, entry):
async def test_fire_transmitter_event(hass, lcn_connection):
"""Test the transmitter event is fired."""
await init_integration(hass, entry)
events = async_capture_events(hass, "lcn_transmitter")
inp = ModStatusAccessControl(
@ -70,7 +55,6 @@ async def test_fire_transmitter_event(hass, entry):
action=KeyAction.HIT,
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -82,11 +66,8 @@ async def test_fire_transmitter_event(hass, entry):
assert events[0].data["action"] == "hit"
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_fire_sendkeys_event(hass, entry):
async def test_fire_sendkeys_event(hass, lcn_connection):
"""Test the send_keys event is fired."""
await init_integration(hass, entry)
events = async_capture_events(hass, "lcn_send_keys")
inp = ModSendKeysHost(
@ -95,7 +76,6 @@ async def test_fire_sendkeys_event(hass, entry):
keys=[True, True, False, False, False, False, False, False],
)
lcn_connection = MockPchkConnectionManager.return_value
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
@ -114,13 +94,9 @@ async def test_fire_sendkeys_event(hass, entry):
assert events[3].data["action"] == "make"
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_dont_fire_on_non_module_input(hass, entry):
async def test_dont_fire_on_non_module_input(hass, lcn_connection):
"""Test for no event is fired if a non-module input is received."""
await init_integration(hass, entry)
inp = Input()
lcn_connection = MockPchkConnectionManager.return_value
for event_name in (
"lcn_transponder",
@ -134,20 +110,15 @@ async def test_dont_fire_on_non_module_input(hass, entry):
assert len(events) == 0
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_dont_fire_on_unknown_module(hass, entry):
async def test_dont_fire_on_unknown_module(hass, lcn_connection):
"""Test for no event is fired if an input from an unknown module is received."""
await init_integration(hass, entry)
inp = ModStatusAccessControl(
LcnAddr(0, 10, False), # unknown module
periphery=AccessControlPeriphery.FINGERPRINT,
code="aabbcc",
)
lcn_connection = MockPchkConnectionManager.return_value
events = async_capture_events(hass, "lcn_transmitter")
events = async_capture_events(hass, "lcn_fingerprint")
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
assert len(events) == 0

47
tests/components/lcn/test_init.py
View File

@ -12,14 +12,11 @@ from homeassistant.components.lcn.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import MockPchkConnectionManager, init_integration, setup_component
from .conftest import MockPchkConnectionManager, setup_component
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_async_setup_entry(hass, entry):
async def test_async_setup_entry(hass, entry, lcn_connection):
"""Test a successful setup entry and unload of entry."""
await init_integration(hass, entry)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.state == ConfigEntryState.LOADED
@ -30,13 +27,14 @@ async def test_async_setup_entry(hass, entry):
assert not hass.data.get(DOMAIN)
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_async_setup_multiple_entries(hass, entry, entry2):
"""Test a successful setup and unload of multiple entries."""
for config_entry in (entry, entry2):
await init_integration(hass, config_entry)
assert config_entry.state == ConfigEntryState.LOADED
await hass.async_block_till_done()
with patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager):
for config_entry in (entry, entry2):
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 2
@ -49,7 +47,6 @@ async def test_async_setup_multiple_entries(hass, entry, entry2):
assert not hass.data.get(DOMAIN)
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_async_setup_entry_update(hass, entry):
"""Test a successful setup entry if entry with same id already exists."""
# setup first entry
@ -74,9 +71,10 @@ async def test_async_setup_entry_update(hass, entry):
assert dummy_device in device_registry.devices.values()
# setup new entry with same data via import step (should cleanup dummy device)
await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=entry.data
)
with patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager):
await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=entry.data
)
assert dummy_device not in device_registry.devices.values()
assert dummy_entity not in entity_registry.entities.values()
@ -87,7 +85,10 @@ async def test_async_setup_entry_raises_authentication_error(hass, entry):
with patch.object(
PchkConnectionManager, "async_connect", side_effect=PchkAuthenticationError
):
await init_integration(hass, entry)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
@ -96,22 +97,28 @@ async def test_async_setup_entry_raises_license_error(hass, entry):
with patch.object(
PchkConnectionManager, "async_connect", side_effect=PchkLicenseError
):
await init_integration(hass, entry)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
async def test_async_setup_entry_raises_timeout_error(hass, entry):
"""Test that an authentication error is handled properly."""
with patch.object(PchkConnectionManager, "async_connect", side_effect=TimeoutError):
await init_integration(hass, entry)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager)
async def test_async_setup_from_configuration_yaml(hass):
"""Test a successful setup using data from configuration.yaml."""
with patch("homeassistant.components.lcn.async_setup_entry") as async_setup_entry:
with patch(
"pypck.connection.PchkConnectionManager", MockPchkConnectionManager
), patch("homeassistant.components.lcn.async_setup_entry") as async_setup_entry:
await setup_component(hass)
assert async_setup_entry.await_count == 2