Migrate calls to async_get_device in rfxtrx tests (#178371)

This commit is contained in:
Erik Montnemery
2026-08-06 19:56:37 +02:00
committed by GitHub
parent f431776656
commit c263da982a
3 changed files with 56 additions and 44 deletions
+25 -19
View File
@@ -22,17 +22,17 @@ class DeviceTestData(NamedTuple):
"""Test data linked to a device."""
code: str
device_identifiers: set[tuple[str, str, str, str]]
device_identifier: tuple[str, str, str, str]
DEVICE_LIGHTING_1 = DeviceTestData("0710002a45050170", {("rfxtrx", "10", "0", "E5")})
DEVICE_LIGHTING_1 = DeviceTestData("0710002a45050170", ("rfxtrx", "10", "0", "E5"))
DEVICE_BLINDS_1 = DeviceTestData(
"09190000009ba8010100", {("rfxtrx", "19", "0", "009ba8:1")}
"09190000009ba8010100", ("rfxtrx", "19", "0", "009ba8:1")
)
DEVICE_TEMPHUM_1 = DeviceTestData(
"0a52080705020095220269", {("rfxtrx", "52", "8", "05:02")}
"0a52080705020095220269", ("rfxtrx", "52", "8", "05:02")
)
@@ -40,12 +40,15 @@ DEVICE_TEMPHUM_1 = DeviceTestData(
async def test_device_test_data(rfxtrx, device: DeviceTestData) -> None:
"""Verify that our testing data remains correct."""
pkt: RFXtrx.lowlevel.Packet = RFXtrx.lowlevel.parse(bytearray.fromhex(device.code))
assert device.device_identifiers == {
("rfxtrx", f"{pkt.packettype:x}", f"{pkt.subtype:x}", pkt.id_string)
}
assert device.device_identifier == (
"rfxtrx",
f"{pkt.packettype:x}",
f"{pkt.subtype:x}",
pkt.id_string,
)
async def setup_entry(hass: HomeAssistant, devices: dict[str, Any]) -> None:
async def setup_entry(hass: HomeAssistant, devices: dict[str, Any]) -> MockConfigEntry:
"""Construct a config setup."""
entry_data = create_rfx_test_cfg(devices=devices)
mock_entry = MockConfigEntry(domain=DOMAIN, unique_id=DOMAIN, data=entry_data)
@@ -56,6 +59,8 @@ async def setup_entry(hass: HomeAssistant, devices: dict[str, Any]) -> None:
await hass.async_block_till_done()
await hass.async_start()
return mock_entry
def _get_expected_actions(data):
for value in data.values():
@@ -83,10 +88,10 @@ async def test_get_actions(
expected,
) -> None:
"""Test we get the expected actions from a rfxtrx."""
await setup_entry(hass, {device.code: {}})
mock_entry = await setup_entry(hass, {device.code: {}})
device_entry = device_registry.async_get_device(
identifiers=device.device_identifiers
device_entry = device_registry.async_get_device_by_identifier(
device.device_identifier, mock_entry.entry_id
)
assert device_entry
@@ -95,8 +100,8 @@ async def test_get_actions(
device_registry.async_update_device(
device_entry.id, merge_identifiers={(identifiers[0], "_".join(identifiers[1:]))}
)
device_entry = device_registry.async_get_device(
identifiers=device.device_identifiers
device_entry = device_registry.async_get_device_by_identifier(
device.device_identifier, mock_entry.entry_id
)
assert device_entry
@@ -143,10 +148,10 @@ async def test_action(
) -> None:
"""Test for actions."""
await setup_entry(hass, {device.code: {}})
mock_entry = await setup_entry(hass, {device.code: {}})
device_entry = device_registry.async_get_device(
identifiers=device.device_identifiers
device_entry = device_registry.async_get_device_by_identifier(
device.device_identifier, mock_entry.entry_id
)
assert device_entry
@@ -184,10 +189,11 @@ async def test_invalid_action(
"""Test for invalid actions."""
device = DEVICE_LIGHTING_1
await setup_entry(hass, {device.code: {}})
mock_entry = await setup_entry(hass, {device.code: {}})
device_identifiers: Any = device.device_identifiers
device_entry = device_registry.async_get_device(identifiers=device_identifiers)
device_entry = device_registry.async_get_device_by_identifier(
device.device_identifier, mock_entry.entry_id
)
assert device_entry
assert await async_setup_component(
+20 -17
View File
@@ -25,26 +25,26 @@ class EventTestData(NamedTuple):
"""Test data linked to a device."""
code: str
device_identifiers: set[tuple[str, str, str, str]]
device_identifier: tuple[str, str, str, str]
type: str
subtype: str
DEVICE_LIGHTING_1 = {("rfxtrx", "10", "0", "E5")}
DEVICE_LIGHTING_1 = ("rfxtrx", "10", "0", "E5")
EVENT_LIGHTING_1 = EventTestData("0710002a45050170", DEVICE_LIGHTING_1, "command", "On")
DEVICE_ROLLERTROL_1 = {("rfxtrx", "19", "0", "009ba8:1")}
DEVICE_ROLLERTROL_1 = ("rfxtrx", "19", "0", "009ba8:1")
EVENT_ROLLERTROL_1 = EventTestData(
"09190000009ba8010100", DEVICE_ROLLERTROL_1, "command", "Down"
)
DEVICE_FIREALARM_1 = {("rfxtrx", "20", "3", "a10900:32")}
DEVICE_FIREALARM_1 = ("rfxtrx", "20", "3", "a10900:32")
EVENT_FIREALARM_1 = EventTestData(
"08200300a109000670", DEVICE_FIREALARM_1, "status", "Panic"
)
async def setup_entry(hass: HomeAssistant, devices: dict[str, Any]) -> None:
async def setup_entry(hass: HomeAssistant, devices: dict[str, Any]) -> MockConfigEntry:
"""Construct a config setup."""
entry_data = create_rfx_test_cfg(devices=devices)
mock_entry = MockConfigEntry(domain=DOMAIN, unique_id=DOMAIN, data=entry_data)
@@ -55,6 +55,8 @@ async def setup_entry(hass: HomeAssistant, devices: dict[str, Any]) -> None:
await hass.async_block_till_done()
await hass.async_start()
return mock_entry
@pytest.mark.parametrize(
("event", "expected"),
@@ -84,20 +86,20 @@ async def test_get_triggers(
expected,
) -> None:
"""Test we get the expected triggers from a rfxtrx."""
await setup_entry(hass, {event.code: {}})
mock_entry = await setup_entry(hass, {event.code: {}})
device_entry = device_registry.async_get_device(
identifiers=event.device_identifiers
device_entry = device_registry.async_get_device_by_identifier(
event.device_identifier, mock_entry.entry_id
)
assert device_entry
# Add alternate identifiers, to make sure we can handle future formats
identifiers: list[str] = list(*event.device_identifiers)
identifiers: list[str] = list(event.device_identifier)
device_registry.async_update_device(
device_entry.id, merge_identifiers={(identifiers[0], "_".join(identifiers[1:]))}
)
device_entry = device_registry.async_get_device(
identifiers=event.device_identifiers
device_entry = device_registry.async_get_device_by_identifier(
event.device_identifier, mock_entry.entry_id
)
assert device_entry
@@ -132,10 +134,10 @@ async def test_firing_event(
) -> None:
"""Test for turn_on and turn_off triggers firing."""
await setup_entry(hass, {event.code: {"fire_event": True}})
mock_entry = await setup_entry(hass, {event.code: {"fire_event": True}})
device_entry = device_registry.async_get_device(
identifiers=event.device_identifiers
device_entry = device_registry.async_get_device_by_identifier(
event.device_identifier, mock_entry.entry_id
)
assert device_entry
@@ -178,10 +180,11 @@ async def test_invalid_trigger(
"""Test for invalid actions."""
event = EVENT_LIGHTING_1
await setup_entry(hass, {event.code: {"fire_event": True}})
mock_entry = await setup_entry(hass, {event.code: {"fire_event": True}})
device_identifiers: Any = event.device_identifiers
device_entry = device_registry.async_get_device(identifiers=device_identifiers)
device_entry = device_registry.async_get_device_by_identifier(
event.device_identifier, mock_entry.entry_id
)
assert device_entry
assert await async_setup_component(
+11 -8
View File
@@ -21,7 +21,7 @@ async def test_fire_event(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, rfxtrx
) -> None:
"""Test fire event."""
await setup_rfx_test_cfg(
mock_entry = await setup_rfx_test_cfg(
hass,
device="/dev/serial/by-id/usb-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
automatic_add=True,
@@ -44,13 +44,13 @@ async def test_fire_event(
await rfxtrx.signal("0b1100cd0213c7f210010f51")
await rfxtrx.signal("0716000100900970")
device_id_1 = device_registry.async_get_device(
identifiers={("rfxtrx", "11", "0", "213c7f2:16")}
device_id_1 = device_registry.async_get_device_by_identifier(
("rfxtrx", "11", "0", "213c7f2:16"), mock_entry.entry_id
)
assert device_id_1
device_id_2 = device_registry.async_get_device(
identifiers={("rfxtrx", "16", "0", "00:90")}
device_id_2 = device_registry.async_get_device_by_identifier(
("rfxtrx", "16", "0", "00:90"), mock_entry.entry_id
)
assert device_id_2
@@ -105,8 +105,8 @@ async def test_ws_device_remove(
},
)
device_entry = device_registry.async_get_device(
identifiers={("rfxtrx", *device_id)}
device_entry = device_registry.async_get_device_by_identifier(
("rfxtrx", *device_id), mock_entry.entry_id
)
assert device_entry
@@ -117,7 +117,10 @@ async def test_ws_device_remove(
# Verify device entry is removed
assert (
device_registry.async_get_device(identifiers={("rfxtrx", *device_id)}) is None
device_registry.async_get_device_by_identifier(
("rfxtrx", *device_id), mock_entry.entry_id
)
is None
)
# Verify that the config entry has removed the device