Add type hints for service_calls fixture in pylint plugin (#118356)

This commit is contained in:
epenet 2024-06-08 16:32:27 +02:00 committed by GitHub
parent b04a65f4d1
commit 5166426d0a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 63 additions and 59 deletions

View File

@ -154,6 +154,7 @@ _TEST_FIXTURES: dict[str, list[str] | str] = {
"recorder_mock": "Recorder",
"request": "pytest.FixtureRequest",
"requests_mock": "Mocker",
"service_calls": "list[ServiceCall]",
"snapshot": "SnapshotAssertion",
"socket_enabled": "None",
"stub_blueprint_populate": "None",

View File

@ -23,11 +23,7 @@ from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
from tests.common import (
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
from tests.common import MockConfigEntry, async_get_device_automations
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -35,12 +31,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.mark.parametrize(
("set_state", "features_reg", "features_state", "expected_condition_types"),
[
@ -189,7 +179,7 @@ async def test_if_state(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
) -> None:
"""Test for all conditions."""
config_entry = MockConfigEntry(domain="test", data={})
@ -373,8 +363,8 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "is_triggered - event - test_event1"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "is_triggered - event - test_event1"
hass.states.async_set(entry.entity_id, STATE_ALARM_DISARMED)
hass.bus.async_fire("test_event1")
@ -385,8 +375,8 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "is_disarmed - event - test_event2"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "is_disarmed - event - test_event2"
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_HOME)
hass.bus.async_fire("test_event1")
@ -397,8 +387,8 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 3
assert calls[2].data["some"] == "is_armed_home - event - test_event3"
assert len(service_calls) == 3
assert service_calls[2].data["some"] == "is_armed_home - event - test_event3"
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_AWAY)
hass.bus.async_fire("test_event1")
@ -409,8 +399,8 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 4
assert calls[3].data["some"] == "is_armed_away - event - test_event4"
assert len(service_calls) == 4
assert service_calls[3].data["some"] == "is_armed_away - event - test_event4"
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_NIGHT)
hass.bus.async_fire("test_event1")
@ -421,8 +411,8 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 5
assert calls[4].data["some"] == "is_armed_night - event - test_event5"
assert len(service_calls) == 5
assert service_calls[4].data["some"] == "is_armed_night - event - test_event5"
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_VACATION)
hass.bus.async_fire("test_event1")
@ -433,8 +423,8 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 6
assert calls[5].data["some"] == "is_armed_vacation - event - test_event6"
assert len(service_calls) == 6
assert service_calls[5].data["some"] == "is_armed_vacation - event - test_event6"
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_CUSTOM_BYPASS)
hass.bus.async_fire("test_event1")
@ -445,15 +435,17 @@ async def test_if_state(
hass.bus.async_fire("test_event6")
hass.bus.async_fire("test_event7")
await hass.async_block_till_done()
assert len(calls) == 7
assert calls[6].data["some"] == "is_armed_custom_bypass - event - test_event7"
assert len(service_calls) == 7
assert (
service_calls[6].data["some"] == "is_armed_custom_bypass - event - test_event7"
)
async def test_if_state_legacy(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
) -> None:
"""Test for all conditions."""
config_entry = MockConfigEntry(domain="test", data={})
@ -499,5 +491,5 @@ async def test_if_state_legacy(
hass.states.async_set(entry.entity_id, STATE_ALARM_TRIGGERED)
hass.bus.async_fire("test_event1")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "is_triggered - event - test_event1"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "is_triggered - event - test_event1"

View File

@ -31,7 +31,6 @@ from tests.common import (
async_fire_time_changed,
async_get_device_automation_capabilities,
async_get_device_automations,
async_mock_service,
)
@ -40,12 +39,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.mark.parametrize(
("set_state", "features_reg", "features_state", "expected_trigger_types"),
[
@ -250,7 +243,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
) -> None:
"""Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={})
@ -409,54 +402,54 @@ async def test_if_fires_on_state_change(
# Fake that the entity is triggered.
hass.states.async_set(entry.entity_id, STATE_ALARM_TRIGGERED)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
assert (
calls[0].data["some"]
service_calls[0].data["some"]
== f"triggered - device - {entry.entity_id} - pending - triggered - None"
)
# Fake that the entity is disarmed.
hass.states.async_set(entry.entity_id, STATE_ALARM_DISARMED)
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
assert (
calls[1].data["some"]
service_calls[1].data["some"]
== f"disarmed - device - {entry.entity_id} - triggered - disarmed - None"
)
# Fake that the entity is armed home.
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_HOME)
await hass.async_block_till_done()
assert len(calls) == 3
assert len(service_calls) == 3
assert (
calls[2].data["some"]
service_calls[2].data["some"]
== f"armed_home - device - {entry.entity_id} - disarmed - armed_home - None"
)
# Fake that the entity is armed away.
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_AWAY)
await hass.async_block_till_done()
assert len(calls) == 4
assert len(service_calls) == 4
assert (
calls[3].data["some"]
service_calls[3].data["some"]
== f"armed_away - device - {entry.entity_id} - armed_home - armed_away - None"
)
# Fake that the entity is armed night.
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_NIGHT)
await hass.async_block_till_done()
assert len(calls) == 5
assert len(service_calls) == 5
assert (
calls[4].data["some"]
service_calls[4].data["some"]
== f"armed_night - device - {entry.entity_id} - armed_away - armed_night - None"
)
# Fake that the entity is armed vacation.
hass.states.async_set(entry.entity_id, STATE_ALARM_ARMED_VACATION)
await hass.async_block_till_done()
assert len(calls) == 6
assert len(service_calls) == 6
assert (
calls[5].data["some"]
service_calls[5].data["some"]
== f"armed_vacation - device - {entry.entity_id} - armed_night - armed_vacation - None"
)
@ -465,7 +458,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
) -> None:
"""Test for triggers firing with delay."""
config_entry = MockConfigEntry(domain="test", data={})
@ -511,17 +504,17 @@ async def test_if_fires_on_state_change_with_for(
},
)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_ALARM_TRIGGERED)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
await hass.async_block_till_done()
assert (
calls[0].data["some"]
service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - disarmed - triggered - 0:00:05"
)
@ -530,7 +523,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
) -> None:
"""Test for triggers firing with delay."""
config_entry = MockConfigEntry(domain="test", data={})
@ -575,12 +568,12 @@ async def test_if_fires_on_state_change_legacy(
},
)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_ALARM_TRIGGERED)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
assert (
calls[0].data["some"]
service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - disarmed - triggered - None"
)

View File

@ -51,7 +51,7 @@ from homeassistant.components.websocket_api.http import URL
from homeassistant.config import YAML_CONFIG_FILE
from homeassistant.config_entries import ConfigEntries, ConfigEntry, ConfigEntryState
from homeassistant.const import HASSIO_USER_NAME
from homeassistant.core import CoreState, HassJob, HomeAssistant
from homeassistant.core import CoreState, HassJob, HomeAssistant, ServiceCall
from homeassistant.helpers import (
area_registry as ar,
category_registry as cr,
@ -1775,6 +1775,24 @@ def label_registry(hass: HomeAssistant) -> lr.LabelRegistry:
return lr.async_get(hass)
@pytest.fixture
def service_calls() -> Generator[None, None, list[ServiceCall]]:
"""Track all service calls."""
calls = []
async def _async_call(
self,
domain: str,
service: str,
service_data: dict[str, Any] | None = None,
**kwargs: Any,
):
calls.append(ServiceCall(domain, service, service_data))
with patch("homeassistant.core.ServiceRegistry.async_call", _async_call):
yield calls
@pytest.fixture
def snapshot(snapshot: SnapshotAssertion) -> SnapshotAssertion:
"""Return snapshot assertion fixture with the Home Assistant extension."""