Use real devices in device_automation device trigger tests (#102684)

This commit is contained in:
Erik Montnemery 2023-10-24 21:57:42 +02:00 committed by GitHub
parent ec3596e85d
commit 4536720540
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -4,12 +4,13 @@ from datetime import timedelta
import pytest
import homeassistant.components.automation as automation
from homeassistant.const import CONF_PLATFORM, STATE_OFF, STATE_ON
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed, async_mock_service
from tests.common import MockConfigEntry, async_fire_time_changed, async_mock_service
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -24,22 +25,27 @@ def calls(hass):
async def test_if_fires_on_state_change(
hass: HomeAssistant, calls, enable_custom_integrations: None
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls,
) -> None:
"""Test for turn_on and turn_off triggers firing.
This is a sanity test for the toggle entity device automation helper, this is
tested by each integration too.
"""
platform = getattr(hass.components, "test.switch")
platform.init()
assert await async_setup_component(
hass, "switch", {"switch": {CONF_PLATFORM: "test"}}
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entry = entity_registry.async_get_or_create(
"switch", "test", "5678", device_id=device_entry.id
)
await hass.async_block_till_done()
ent1, ent2, ent3 = platform.ENTITIES
hass.states.async_set(entry.entity_id, STATE_ON)
assert await async_setup_component(
hass,
@ -50,8 +56,8 @@ async def test_if_fires_on_state_change(
"trigger": {
"platform": "device",
"domain": "switch",
"device_id": "",
"entity_id": ent1.entity_id,
"device_id": device_entry.id,
"entity_id": entry.entity_id,
"type": "turned_on",
},
"action": {
@ -74,8 +80,8 @@ async def test_if_fires_on_state_change(
"trigger": {
"platform": "device",
"domain": "switch",
"device_id": "",
"entity_id": ent1.entity_id,
"device_id": device_entry.id,
"entity_id": entry.entity_id,
"type": "turned_off",
},
"action": {
@ -98,8 +104,8 @@ async def test_if_fires_on_state_change(
"trigger": {
"platform": "device",
"domain": "switch",
"device_id": "",
"entity_id": ent1.entity_id,
"device_id": device_entry.id,
"entity_id": entry.entity_id,
"type": "changed_states",
},
"action": {
@ -122,40 +128,46 @@ async def test_if_fires_on_state_change(
},
)
await hass.async_block_till_done()
assert hass.states.get(ent1.entity_id).state == STATE_ON
assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0
hass.states.async_set(ent1.entity_id, STATE_OFF)
hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done()
assert len(calls) == 2
assert {calls[0].data["some"], calls[1].data["some"]} == {
f"turn_off device - {ent1.entity_id} - on - off - None",
f"turn_on_or_off device - {ent1.entity_id} - on - off - None",
f"turn_off device - {entry.entity_id} - on - off - None",
f"turn_on_or_off device - {entry.entity_id} - on - off - None",
}
hass.states.async_set(ent1.entity_id, STATE_ON)
hass.states.async_set(entry.entity_id, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 4
assert {calls[2].data["some"], calls[3].data["some"]} == {
f"turn_on device - {ent1.entity_id} - off - on - None",
f"turn_on_or_off device - {ent1.entity_id} - off - on - None",
f"turn_on device - {entry.entity_id} - off - on - None",
f"turn_on_or_off device - {entry.entity_id} - off - on - None",
}
@pytest.mark.parametrize("trigger", ["turned_off", "changed_states"])
async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, calls, enable_custom_integrations: None, trigger
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls,
trigger,
) -> None:
"""Test for triggers firing with delay."""
platform = getattr(hass.components, "test.switch")
platform.init()
assert await async_setup_component(
hass, "switch", {"switch": {CONF_PLATFORM: "test"}}
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entry = entity_registry.async_get_or_create(
"switch", "test", "5678", device_id=device_entry.id
)
await hass.async_block_till_done()
ent1, ent2, ent3 = platform.ENTITIES
hass.states.async_set(entry.entity_id, STATE_ON)
assert await async_setup_component(
hass,
@ -166,8 +178,8 @@ async def test_if_fires_on_state_change_with_for(
"trigger": {
"platform": "device",
"domain": "switch",
"device_id": "",
"entity_id": ent1.entity_id,
"device_id": device_entry.id,
"entity_id": entry.entity_id,
"type": trigger,
"for": {"seconds": 5},
},
@ -191,10 +203,10 @@ async def test_if_fires_on_state_change_with_for(
},
)
await hass.async_block_till_done()
assert hass.states.get(ent1.entity_id).state == STATE_ON
assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0
hass.states.async_set(ent1.entity_id, STATE_OFF)
hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done()
assert len(calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
@ -202,5 +214,5 @@ async def test_if_fires_on_state_change_with_for(
assert len(calls) == 1
await hass.async_block_till_done()
assert calls[0].data["some"] == "turn_off device - {} - on - off - 0:00:05".format(
ent1.entity_id
entry.entity_id
)