150 lines
4.5 KiB
Python
150 lines
4.5 KiB
Python
"""Shared fixtures for Main Gate Controller tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Generator
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from homeassistant.core import HomeAssistant, ServiceCall
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.main_gate_controller.const import (
|
|
CONF_NAME,
|
|
CONF_NOTIFY_CLOSING_TEXT,
|
|
CONF_NOTIFY_ENABLED,
|
|
CONF_NOTIFY_ENTITIES,
|
|
CONF_NOTIFY_OPENING_TEXT,
|
|
CONF_OPEN_DURATION,
|
|
CONF_PULSE_DURATION,
|
|
CONF_SWITCH_ENTITY_ID,
|
|
DEFAULT_NAME,
|
|
DEFAULT_NOTIFY_CLOSING_TEXT,
|
|
DEFAULT_NOTIFY_ENABLED,
|
|
DEFAULT_NOTIFY_OPENING_TEXT,
|
|
DEFAULT_OPEN_DURATION,
|
|
DEFAULT_PULSE_DURATION,
|
|
DOMAIN,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def auto_enable_custom_integrations(enable_custom_integrations: None) -> Generator[None]:
|
|
"""Enable the bundled custom integration via PHACC's loader patch."""
|
|
yield
|
|
|
|
|
|
@dataclass
|
|
class ServiceCalls:
|
|
"""Helper to inspect service call recordings in a structured way."""
|
|
|
|
records: list[tuple[str, str, dict[str, Any]]] = field(default_factory=list)
|
|
|
|
def service(self, domain: str, service: str) -> list[dict[str, Any]]:
|
|
return [data for d, s, data in self.records if d == domain and s == service]
|
|
|
|
def order(self) -> list[tuple[str, str]]:
|
|
return [(d, s) for d, s, _ in self.records]
|
|
|
|
def reset(self) -> None:
|
|
self.records.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def service_calls(hass: HomeAssistant) -> ServiceCalls:
|
|
"""Register recording handlers for `switch` and `notify` services."""
|
|
|
|
helper = ServiceCalls()
|
|
|
|
async def _record(call: ServiceCall) -> None:
|
|
helper.records.append((call.domain, call.service, dict(call.data)))
|
|
|
|
hass.services.async_register("switch", "turn_on", _record)
|
|
hass.services.async_register("switch", "turn_off", _record)
|
|
hass.services.async_register("notify", "send_message", _record)
|
|
return helper
|
|
|
|
|
|
@pytest.fixture
|
|
def gate_switch(hass: HomeAssistant) -> str:
|
|
"""Provide a real-looking switch entity in the state machine."""
|
|
entity_id = "switch.main_gate_relay"
|
|
hass.states.async_set(entity_id, "off")
|
|
return entity_id
|
|
|
|
|
|
@pytest.fixture
|
|
def notify_entity(hass: HomeAssistant) -> str:
|
|
"""Provide a notify entity in the state machine (modern Notify entity platform)."""
|
|
entity_id = "notify.mobile_app_phone"
|
|
hass.states.async_set(entity_id, "unknown")
|
|
return entity_id
|
|
|
|
|
|
def make_entry_data(
|
|
*,
|
|
switch_entity_id: str,
|
|
name: str = DEFAULT_NAME,
|
|
open_duration: float = DEFAULT_OPEN_DURATION,
|
|
pulse_duration: float = DEFAULT_PULSE_DURATION,
|
|
notify_enabled: bool = DEFAULT_NOTIFY_ENABLED,
|
|
notify_entities: list[str] | None = None,
|
|
notify_opening_text: str = DEFAULT_NOTIFY_OPENING_TEXT,
|
|
notify_closing_text: str = DEFAULT_NOTIFY_CLOSING_TEXT,
|
|
) -> dict[str, Any]:
|
|
"""Build a default config-entry data dict."""
|
|
return {
|
|
CONF_NAME: name,
|
|
CONF_SWITCH_ENTITY_ID: switch_entity_id,
|
|
CONF_OPEN_DURATION: open_duration,
|
|
CONF_PULSE_DURATION: pulse_duration,
|
|
CONF_NOTIFY_ENABLED: notify_enabled,
|
|
CONF_NOTIFY_ENTITIES: notify_entities or [],
|
|
CONF_NOTIFY_OPENING_TEXT: notify_opening_text,
|
|
CONF_NOTIFY_CLOSING_TEXT: notify_closing_text,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def config_flow_user_input(gate_switch: str, notify_entity: str) -> dict[str, Any]:
|
|
"""User input ready to be submitted to the config flow."""
|
|
return make_entry_data(
|
|
switch_entity_id=gate_switch,
|
|
name="Main Gate",
|
|
open_duration=DEFAULT_OPEN_DURATION,
|
|
pulse_duration=DEFAULT_PULSE_DURATION,
|
|
notify_enabled=False,
|
|
notify_entities=[],
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def loaded_entry(
|
|
hass: HomeAssistant,
|
|
gate_switch: str,
|
|
notify_entity: str | None,
|
|
) -> MockConfigEntry:
|
|
"""Provide a fully set up config entry with two entities."""
|
|
if notify_entity is None:
|
|
notify_entities: list[str] = []
|
|
else:
|
|
notify_entities = [notify_entity]
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
title="Main Gate",
|
|
data=make_entry_data(
|
|
switch_entity_id=gate_switch,
|
|
name="Main Gate",
|
|
open_duration=20,
|
|
pulse_duration=1.0,
|
|
notify_enabled=bool(notify_entities),
|
|
notify_entities=notify_entities,
|
|
),
|
|
unique_id=gate_switch,
|
|
)
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
return entry
|