Use service_calls fixture in zwave_js tests (#120994)

pull/121012/head
epenet 2024-07-02 12:37:21 +02:00 committed by GitHub
parent 02dffcde1a
commit 3adea1ada9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 56 additions and 68 deletions

View File

@ -25,13 +25,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, device_registry as dr from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import async_get_device_automations, async_mock_service from tests.common import async_get_device_automations
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions( async def test_get_conditions(
@ -99,7 +93,7 @@ async def test_node_status_state(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
) -> None: ) -> None:
"""Test for node_status conditions.""" """Test for node_status conditions."""
@ -206,8 +200,8 @@ async def test_node_status_state(
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4") hass.bus.async_fire("test_event4")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "alive - event - test_event1" assert service_calls[0].data["some"] == "alive - event - test_event1"
event = Event( event = Event(
"wake up", "wake up",
@ -225,8 +219,8 @@ async def test_node_status_state(
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4") hass.bus.async_fire("test_event4")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "awake - event - test_event2" assert service_calls[1].data["some"] == "awake - event - test_event2"
event = Event( event = Event(
"sleep", "sleep",
@ -240,8 +234,8 @@ async def test_node_status_state(
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4") hass.bus.async_fire("test_event4")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[2].data["some"] == "asleep - event - test_event3" assert service_calls[2].data["some"] == "asleep - event - test_event3"
event = Event( event = Event(
"dead", "dead",
@ -255,8 +249,8 @@ async def test_node_status_state(
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4") hass.bus.async_fire("test_event4")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert calls[3].data["some"] == "dead - event - test_event4" assert service_calls[3].data["some"] == "dead - event - test_event4"
async def test_config_parameter_state( async def test_config_parameter_state(
@ -264,7 +258,7 @@ async def test_config_parameter_state(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
) -> None: ) -> None:
"""Test for config_parameter conditions.""" """Test for config_parameter conditions."""
@ -331,8 +325,8 @@ async def test_config_parameter_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "Beeper - event - test_event1" assert service_calls[0].data["some"] == "Beeper - event - test_event1"
# Flip Beeper state to not match condition # Flip Beeper state to not match condition
event = Event( event = Event(
@ -375,8 +369,8 @@ async def test_config_parameter_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "User Slot Status - event - test_event2" assert service_calls[1].data["some"] == "User Slot Status - event - test_event2"
async def test_value_state( async def test_value_state(
@ -384,7 +378,7 @@ async def test_value_state(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
) -> None: ) -> None:
"""Test for value conditions.""" """Test for value conditions."""
@ -427,8 +421,8 @@ async def test_value_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "value - event - test_event1" assert service_calls[0].data["some"] == "value - event - test_event1"
async def test_get_condition_capabilities_node_status( async def test_get_condition_capabilities_node_status(

View File

@ -28,13 +28,7 @@ from homeassistant.helpers import (
) )
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import async_get_device_automations, async_mock_service from tests.common import async_get_device_automations
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_no_controller_triggers( async def test_no_controller_triggers(
@ -85,7 +79,7 @@ async def test_if_notification_notification_fires(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for event.notification.notification trigger firing.""" """Test for event.notification.notification trigger firing."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -168,13 +162,13 @@ async def test_if_notification_notification_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"event.notification.notification - device - zwave_js_notification - {CommandClass.NOTIFICATION}" == f"event.notification.notification - device - zwave_js_notification - {CommandClass.NOTIFICATION}"
) )
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"event.notification.notification2 - device - zwave_js_notification - {CommandClass.NOTIFICATION}" == f"event.notification.notification2 - device - zwave_js_notification - {CommandClass.NOTIFICATION}"
) )
@ -221,7 +215,7 @@ async def test_if_entry_control_notification_fires(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for notification.entry_control trigger firing.""" """Test for notification.entry_control trigger firing."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -303,13 +297,13 @@ async def test_if_entry_control_notification_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"event.notification.notification - device - zwave_js_notification - {CommandClass.ENTRY_CONTROL}" == f"event.notification.notification - device - zwave_js_notification - {CommandClass.ENTRY_CONTROL}"
) )
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"event.notification.notification2 - device - zwave_js_notification - {CommandClass.ENTRY_CONTROL}" == f"event.notification.notification2 - device - zwave_js_notification - {CommandClass.ENTRY_CONTROL}"
) )
@ -389,7 +383,7 @@ async def test_if_node_status_change_fires(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for node_status trigger firing.""" """Test for node_status trigger firing."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -460,9 +454,9 @@ async def test_if_node_status_change_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[0].data["some"] == "state.node_status - device - alive" assert service_calls[0].data["some"] == "state.node_status - device - alive"
assert calls[1].data["some"] == "state.node_status2 - device - alive" assert service_calls[1].data["some"] == "state.node_status2 - device - alive"
async def test_if_node_status_change_fires_legacy( async def test_if_node_status_change_fires_legacy(
@ -472,7 +466,7 @@ async def test_if_node_status_change_fires_legacy(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for node_status trigger firing.""" """Test for node_status trigger firing."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -543,9 +537,9 @@ async def test_if_node_status_change_fires_legacy(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[0].data["some"] == "state.node_status - device - alive" assert service_calls[0].data["some"] == "state.node_status - device - alive"
assert calls[1].data["some"] == "state.node_status2 - device - alive" assert service_calls[1].data["some"] == "state.node_status2 - device - alive"
async def test_get_trigger_capabilities_node_status( async def test_get_trigger_capabilities_node_status(
@ -645,7 +639,7 @@ async def test_if_basic_value_notification_fires(
client, client,
ge_in_wall_dimmer_switch, ge_in_wall_dimmer_switch,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for event.value_notification.basic trigger firing.""" """Test for event.value_notification.basic trigger firing."""
node: Node = ge_in_wall_dimmer_switch node: Node = ge_in_wall_dimmer_switch
@ -742,13 +736,13 @@ async def test_if_basic_value_notification_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"event.value_notification.basic - device - zwave_js_value_notification - {CommandClass.BASIC}" == f"event.value_notification.basic - device - zwave_js_value_notification - {CommandClass.BASIC}"
) )
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"event.value_notification.basic2 - device - zwave_js_value_notification - {CommandClass.BASIC}" == f"event.value_notification.basic2 - device - zwave_js_value_notification - {CommandClass.BASIC}"
) )
@ -830,7 +824,7 @@ async def test_if_central_scene_value_notification_fires(
client, client,
wallmote_central_scene, wallmote_central_scene,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for event.value_notification.central_scene trigger firing.""" """Test for event.value_notification.central_scene trigger firing."""
node: Node = wallmote_central_scene node: Node = wallmote_central_scene
@ -933,13 +927,13 @@ async def test_if_central_scene_value_notification_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"event.value_notification.central_scene - device - zwave_js_value_notification - {CommandClass.CENTRAL_SCENE}" == f"event.value_notification.central_scene - device - zwave_js_value_notification - {CommandClass.CENTRAL_SCENE}"
) )
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"event.value_notification.central_scene2 - device - zwave_js_value_notification - {CommandClass.CENTRAL_SCENE}" == f"event.value_notification.central_scene2 - device - zwave_js_value_notification - {CommandClass.CENTRAL_SCENE}"
) )
@ -1020,7 +1014,7 @@ async def test_if_scene_activation_value_notification_fires(
client, client,
hank_binary_switch, hank_binary_switch,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for event.value_notification.scene_activation trigger firing.""" """Test for event.value_notification.scene_activation trigger firing."""
node: Node = hank_binary_switch node: Node = hank_binary_switch
@ -1117,13 +1111,13 @@ async def test_if_scene_activation_value_notification_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"event.value_notification.scene_activation - device - zwave_js_value_notification - {CommandClass.SCENE_ACTIVATION}" == f"event.value_notification.scene_activation - device - zwave_js_value_notification - {CommandClass.SCENE_ACTIVATION}"
) )
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"event.value_notification.scene_activation2 - device - zwave_js_value_notification - {CommandClass.SCENE_ACTIVATION}" == f"event.value_notification.scene_activation2 - device - zwave_js_value_notification - {CommandClass.SCENE_ACTIVATION}"
) )
@ -1200,7 +1194,7 @@ async def test_if_value_updated_value_fires(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for zwave_js.value_updated.value trigger firing.""" """Test for zwave_js.value_updated.value trigger firing."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -1261,7 +1255,7 @@ async def test_if_value_updated_value_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
# Publish fake value update that should trigger # Publish fake value update that should trigger
event = Event( event = Event(
@ -1283,9 +1277,9 @@ async def test_if_value_updated_value_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== "zwave_js.value_updated.value - zwave_js.value_updated - open" == "zwave_js.value_updated.value - zwave_js.value_updated - open"
) )
@ -1296,7 +1290,7 @@ async def test_value_updated_value_no_driver(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test zwave_js.value_updated.value trigger with missing driver.""" """Test zwave_js.value_updated.value trigger with missing driver."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -1362,7 +1356,7 @@ async def test_value_updated_value_no_driver(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
async def test_get_trigger_capabilities_value_updated_value( async def test_get_trigger_capabilities_value_updated_value(
@ -1455,7 +1449,7 @@ async def test_if_value_updated_config_parameter_fires(
client, client,
lock_schlage_be469, lock_schlage_be469,
integration, integration,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for zwave_js.value_updated.config_parameter trigger firing.""" """Test for zwave_js.value_updated.config_parameter trigger firing."""
node: Node = lock_schlage_be469 node: Node = lock_schlage_be469
@ -1517,9 +1511,9 @@ async def test_if_value_updated_config_parameter_fires(
) )
node.receive_event(event) node.receive_event(event)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== "zwave_js.value_updated.config_parameter - zwave_js.value_updated - 255" == "zwave_js.value_updated.config_parameter - zwave_js.value_updated - 255"
) )