Use service_calls fixture in arcam_fmj tests (#119274)

This commit is contained in:
epenet
2024-06-10 13:09:51 +02:00
committed by GitHub
parent b8e57f6174
commit 8cbfc5a58b
2 changed files with 30 additions and 22 deletions

View File

@@ -9,11 +9,7 @@ from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@@ -21,12 +17,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """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")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@@ -69,7 +59,7 @@ async def test_get_triggers(
async def test_if_fires_on_turn_on_request( async def test_if_fires_on_turn_on_request(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
player_setup, player_setup,
state, state,
) -> None: ) -> None:
@@ -111,15 +101,15 @@ async def test_if_fires_on_turn_on_request(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 2
assert calls[0].data["some"] == player_setup assert service_calls[1].data["some"] == player_setup
assert calls[0].data["id"] == 0 assert service_calls[1].data["id"] == 0
async def test_if_fires_on_turn_on_request_legacy( async def test_if_fires_on_turn_on_request_legacy(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
player_setup, player_setup,
state, state,
) -> None: ) -> None:
@@ -161,6 +151,6 @@ async def test_if_fires_on_turn_on_request_legacy(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 2
assert calls[0].data["some"] == player_setup assert service_calls[1].data["some"] == player_setup
assert calls[0].data["id"] == 0 assert service_calls[1].data["id"] == 0

View File

@@ -51,7 +51,13 @@ from homeassistant.components.websocket_api.http import URL
from homeassistant.config import YAML_CONFIG_FILE from homeassistant.config import YAML_CONFIG_FILE
from homeassistant.config_entries import ConfigEntries, ConfigEntry, ConfigEntryState from homeassistant.config_entries import ConfigEntries, ConfigEntry, ConfigEntryState
from homeassistant.const import HASSIO_USER_NAME from homeassistant.const import HASSIO_USER_NAME
from homeassistant.core import CoreState, HassJob, HomeAssistant, ServiceCall from homeassistant.core import (
CoreState,
HassJob,
HomeAssistant,
ServiceCall,
ServiceResponse,
)
from homeassistant.helpers import ( from homeassistant.helpers import (
area_registry as ar, area_registry as ar,
category_registry as cr, category_registry as cr,
@@ -1776,18 +1782,30 @@ def label_registry(hass: HomeAssistant) -> lr.LabelRegistry:
@pytest.fixture @pytest.fixture
def service_calls() -> Generator[None, None, list[ServiceCall]]: def service_calls(hass: HomeAssistant) -> Generator[None, None, list[ServiceCall]]:
"""Track all service calls.""" """Track all service calls."""
calls = [] calls = []
_original_async_call = hass.services.async_call
async def _async_call( async def _async_call(
self, self,
domain: str, domain: str,
service: str, service: str,
service_data: dict[str, Any] | None = None, service_data: dict[str, Any] | None = None,
**kwargs: Any, **kwargs: Any,
): ) -> ServiceResponse:
calls.append(ServiceCall(domain, service, service_data)) calls.append(ServiceCall(domain, service, service_data))
try:
return await _original_async_call(
domain,
service,
service_data,
**kwargs,
)
except ha.ServiceNotFound:
_LOGGER.debug("Ignoring unknown service call to %s.%s", domain, service)
return None
with patch("homeassistant.core.ServiceRegistry.async_call", _async_call): with patch("homeassistant.core.ServiceRegistry.async_call", _async_call):
yield calls yield calls