evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use service_calls fixture in knx tests (#120930)

Browse source
This commit is contained in:
epenet 2024-07-01 14:33:01 +02:00 • committed by GitHub
parent 5513682de4
commit bc3562a9e8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 58 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
tests/components/knx/test_device_trigger.py
View file

@ -18,18 +18,12 @@ from homeassistant.setup import async_setup_component
from .conftest import KNXTestKit
from tests.common import async_get_device_automations, async_mock_service
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
from tests.common import async_get_device_automations
async def test_if_fires_on_telegram(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
knx: KNXTestKit,
) -> None:
@ -98,31 +92,31 @@ async def test_if_fires_on_telegram(
# "specific" shall ignore destination address
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["id"] == 0
await knx.receive_write("1/2/4", (0x03, 0x2F))
assert len(calls) == 2
test_call = calls.pop()
assert len(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = calls.pop()
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
# "specific" shall ignore GroupValueRead
await knx.receive_read("1/2/4")
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
async def test_default_if_fires_on_telegram(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
knx: KNXTestKit,
) -> None:
@ -179,34 +173,34 @@ async def test_default_if_fires_on_telegram(
)
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["id"] == 0
await knx.receive_write("1/2/4", (0x03, 0x2F))
assert len(calls) == 2
test_call = calls.pop()
assert len(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = calls.pop()
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
# "specific" shall catch GroupValueRead as it is not set explicitly
await knx.receive_read("1/2/4")
assert len(calls) == 2
test_call = calls.pop()
assert len(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = calls.pop()
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
async def test_remove_device_trigger(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
knx: KNXTestKit,
) -> None:
@ -241,8 +235,8 @@ async def test_remove_device_trigger(
)
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
assert len(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
await hass.services.async_call(
automation.DOMAIN,
@ -250,8 +244,10 @@ async def test_remove_device_trigger(
{ATTR_ENTITY_ID: f"automation.{automation_name}"},
blocking=True,
)
assert len(service_calls) == 1
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 0
assert len(service_calls) == 1
async def test_get_triggers(

72
tests/components/knx/test_trigger.py
View file

@ -11,18 +11,10 @@ from homeassistant.setup import async_setup_component
from .conftest import KNXTestKit
from tests.common import async_mock_service
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_telegram_trigger(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
knx: KNXTestKit,
) -> None:
"""Test telegram triggers firing."""
@ -73,24 +65,24 @@ async def test_telegram_trigger(
# "specific" shall ignore destination address
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["id"] == 0
await knx.receive_write("1/2/4", (0x03, 0x2F))
assert len(calls) == 2
test_call = calls.pop()
assert len(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = calls.pop()
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
# "specific" shall ignore GroupValueRead
await knx.receive_read("1/2/4")
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
@ -105,7 +97,7 @@ async def test_telegram_trigger(
)
async def test_telegram_trigger_dpt_option(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
knx: KNXTestKit,
payload: tuple[int, ...],
type_option: dict[str, bool],
@ -138,16 +130,16 @@ async def test_telegram_trigger_dpt_option(
)
await knx.receive_write("0/0/1", payload)
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["trigger"]["value"] == expected_value
assert test_call.data["trigger"]["unit"] == expected_unit
await knx.receive_read("0/0/1")
assert len(calls) == 1
test_call = calls.pop()
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["trigger"]["value"] is None
assert test_call.data["trigger"]["unit"] is None
@ -192,7 +184,7 @@ async def test_telegram_trigger_dpt_option(
)
async def test_telegram_trigger_options(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
knx: KNXTestKit,
group_value_options: dict[str, bool],
direction_options: dict[str, bool],
@ -225,28 +217,28 @@ async def test_telegram_trigger_options(
if group_value_options.get("group_value_write", True) and direction_options.get(
"incoming", True
):
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
assert len(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(calls) == 0
assert len(service_calls) == 0
await knx.receive_response("0/0/1", 1)
if group_value_options["group_value_response"] and direction_options.get(
"incoming", True
):
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
assert len(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(calls) == 0
assert len(service_calls) == 0
await knx.receive_read("0/0/1")
if group_value_options["group_value_read"] and direction_options.get(
"incoming", True
):
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
assert len(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(calls) == 0
assert len(service_calls) == 0
await hass.services.async_call(
"knx",
@ -254,20 +246,22 @@ async def test_telegram_trigger_options(
{"address": "0/0/1", "payload": True},
blocking=True,
)
assert len(service_calls) == 1
await knx.assert_write("0/0/1", True)
if (
group_value_options.get("group_value_write", True)
and direction_options["outgoing"]
):
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
assert len(service_calls) == 2
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(calls) == 0
assert len(service_calls) == 1
async def test_remove_telegram_trigger(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
knx: KNXTestKit,
) -> None:
"""Test for removed callback when telegram trigger not used."""
@ -296,8 +290,8 @@ async def test_remove_telegram_trigger(
)
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
assert len(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
await hass.services.async_call(
automation.DOMAIN,
@ -305,8 +299,10 @@ async def test_remove_telegram_trigger(
{ATTR_ENTITY_ID: f"automation.{automation_name}"},
blocking=True,
)
assert len(service_calls) == 1
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 0
assert len(service_calls) == 1
async def test_invalid_trigger(