Initial commit: Main Gate Controller HACS integration
This commit is contained in:
bot
@@ -0,0 +1,149 @@
|
||||
"""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
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Tests for the open/countdown/close sequence driven by the button."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.const import SERVICE_TURN_OFF, SERVICE_TURN_ON
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.main_gate_controller.const import (
|
||||
DOMAIN,
|
||||
GATE_STATE_CLOSED,
|
||||
GATE_STATE_CLOSING,
|
||||
GATE_STATE_OPEN,
|
||||
GATE_STATE_OPENING,
|
||||
NOTIFY_DOMAIN,
|
||||
SERVICE_SEND_MESSAGE,
|
||||
SWITCH_DOMAIN,
|
||||
)
|
||||
from tests.conftest import ServiceCalls, make_entry_data
|
||||
|
||||
|
||||
async def _press_button(hass: HomeAssistant, entry) -> None:
|
||||
registry = er.async_get(hass)
|
||||
entity_id = registry.async_get_entity_id(
|
||||
"button", DOMAIN, f"{entry.entry_id}_open"
|
||||
)
|
||||
assert entity_id is not None
|
||||
await hass.services.async_call(
|
||||
"button", "press", {"entity_id": entity_id}, blocking=True
|
||||
)
|
||||
|
||||
|
||||
async def test_full_open_close_sequence_with_notifications(
|
||||
service_calls: ServiceCalls, loaded_entry
|
||||
):
|
||||
"""Notifications-enabled entry runs the canonical sequence.
|
||||
|
||||
The spec dictates: ``Closing`` notification fires **before** the closing
|
||||
switch pulses (state-machine transition closing -> closed is logged from
|
||||
the coordinator inside the switch sequence).
|
||||
"""
|
||||
controller = loaded_entry.runtime_data
|
||||
assert controller.async_trigger() is True
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
expected = [
|
||||
(NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE), # Opening notification
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_OFF), # OPEN: OFF
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_ON), # OPEN: ON
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_OFF), # OPEN: OFF
|
||||
(NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE), # Closing notification (before close pulse)
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_ON), # CLOSE: ON
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_OFF), # CLOSE: OFF
|
||||
]
|
||||
assert service_calls.order() == expected
|
||||
|
||||
notifies = service_calls.service(NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE)
|
||||
assert notifies[0]["message"].startswith("Opening Main Gate")
|
||||
assert notifies[1]["message"] == "Closing Main Gate"
|
||||
|
||||
targets = notifies[0]["target"]
|
||||
if isinstance(targets, str):
|
||||
targets = [targets]
|
||||
assert targets == ["notify.mobile_app_phone"]
|
||||
|
||||
assert controller.status == GATE_STATE_CLOSED
|
||||
assert controller.is_running is False
|
||||
|
||||
|
||||
async def test_sequence_without_notifications(
|
||||
service_calls: ServiceCalls, hass: HomeAssistant, gate_switch: str
|
||||
):
|
||||
"""Notifications disabled → only switch calls happen."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(
|
||||
switch_entity_id=gate_switch, name="Main Gate", notify_enabled=False
|
||||
),
|
||||
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()
|
||||
|
||||
controller = entry.runtime_data
|
||||
assert controller.async_trigger() is True
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
expected = [
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_OFF),
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_ON),
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_OFF),
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_ON),
|
||||
(SWITCH_DOMAIN, SERVICE_TURN_OFF),
|
||||
]
|
||||
assert service_calls.order() == expected
|
||||
assert controller.status == GATE_STATE_CLOSED
|
||||
|
||||
|
||||
async def test_mode_single_ignores_second_trigger(
|
||||
service_calls: ServiceCalls, loaded_entry
|
||||
):
|
||||
"""A second trigger during a running cycle does nothing (mode: single)."""
|
||||
controller = loaded_entry.runtime_data
|
||||
first = controller.async_trigger()
|
||||
second = controller.async_trigger()
|
||||
assert first is True
|
||||
assert second is False
|
||||
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
switch_calls = sum(
|
||||
1 for d, s, _ in service_calls.records if d == SWITCH_DOMAIN
|
||||
)
|
||||
assert switch_calls == 5
|
||||
|
||||
|
||||
async def test_button_press_starts_cycle(
|
||||
service_calls: ServiceCalls, hass: HomeAssistant, loaded_entry
|
||||
):
|
||||
"""Pressing the button entity starts the cycle via the service call."""
|
||||
controller = loaded_entry.runtime_data
|
||||
await _press_button(hass, loaded_entry)
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
switch_calls = sum(
|
||||
1 for d, s, _ in service_calls.records if d == SWITCH_DOMAIN
|
||||
)
|
||||
# 5 switch calls (no notify because loaded_entry is without notify targets).
|
||||
assert switch_calls == 5
|
||||
|
||||
|
||||
async def test_state_transitions_during_cycle(
|
||||
hass: HomeAssistant, service_calls: ServiceCalls, loaded_entry
|
||||
):
|
||||
"""The status sensor goes opening → open → closing → closed."""
|
||||
# Subscribe to state changes of the status sensor.
|
||||
observed: list[str] = []
|
||||
|
||||
def _capture(event):
|
||||
if event.data.get("entity_id", "").endswith("_status"):
|
||||
new_state: dict[str, Any] = event.data.get("new_state")
|
||||
if new_state is not None and new_state.state in {
|
||||
GATE_STATE_CLOSED,
|
||||
GATE_STATE_OPENING,
|
||||
GATE_STATE_OPEN,
|
||||
GATE_STATE_CLOSING,
|
||||
}:
|
||||
if not observed or observed[-1] != new_state.state:
|
||||
observed.append(new_state.state)
|
||||
|
||||
hass.bus.async_listen("state_changed", _capture)
|
||||
|
||||
controller = loaded_entry.runtime_data
|
||||
assert controller.async_trigger() is True
|
||||
await controller.async_wait_until_idle()
|
||||
# Let pending state_changed events propagate through the bus.
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Opening and closing are short with pulse=1s; ``open`` is the dominant state.
|
||||
seen = set(observed)
|
||||
assert GATE_STATE_OPENING in seen
|
||||
assert GATE_STATE_OPEN in seen
|
||||
assert GATE_STATE_CLOSING in seen
|
||||
assert GATE_STATE_CLOSED in seen
|
||||
@@ -0,0 +1,131 @@
|
||||
"""Tests for the user-initiated config flow."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from custom_components.main_gate_controller.const import (
|
||||
CONF_NAME,
|
||||
CONF_NOTIFY_ENABLED,
|
||||
CONF_NOTIFY_ENTITIES,
|
||||
CONF_SWITCH_ENTITY_ID,
|
||||
DOMAIN,
|
||||
)
|
||||
from tests.conftest import make_entry_data
|
||||
|
||||
|
||||
async def _start_user_flow(hass: HomeAssistant) -> str:
|
||||
"""Initiate the user config flow and return its flow_id."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
return result["flow_id"]
|
||||
|
||||
|
||||
async def test_user_flow_creates_entry(
|
||||
hass: HomeAssistant, gate_switch: str, notify_entity: str
|
||||
):
|
||||
"""Valid input creates a config entry."""
|
||||
flow_id = await _start_user_flow(hass)
|
||||
user_input = make_entry_data(
|
||||
switch_entity_id=gate_switch,
|
||||
name="Main Gate",
|
||||
open_duration=20,
|
||||
pulse_duration=1.0,
|
||||
notify_enabled=False,
|
||||
notify_entities=[],
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Main Gate"
|
||||
assert result["data"][CONF_SWITCH_ENTITY_ID] == gate_switch
|
||||
assert result["data"][CONF_NAME] == "Main Gate"
|
||||
|
||||
|
||||
async def test_duplicate_switch_aborts(hass: HomeAssistant, gate_switch: str):
|
||||
"""Setting up the same switch twice aborts with ``already_configured``."""
|
||||
flow_id = await _start_user_flow(hass)
|
||||
user_input = make_entry_data(switch_entity_id=gate_switch, name="Gate A")
|
||||
first = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
assert first["type"] == FlowResultType.CREATE_ENTRY
|
||||
|
||||
# Start a second flow with the same switch.
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
second = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
assert second["type"] == FlowResultType.ABORT
|
||||
assert second["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_missing_notify_targets_shows_error(
|
||||
hass: HomeAssistant, gate_switch: str, notify_entity: str
|
||||
):
|
||||
"""Enabling notifications without targets reports a validation error."""
|
||||
flow_id = await _start_user_flow(hass)
|
||||
user_input = make_entry_data(
|
||||
switch_entity_id=gate_switch,
|
||||
name="Main Gate",
|
||||
notify_enabled=True,
|
||||
notify_entities=[],
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_NOTIFY_ENTITIES: "notify_no_targets"}
|
||||
|
||||
|
||||
async def test_unknown_switch_entity_shows_error(
|
||||
hass: HomeAssistant, gate_switch: str
|
||||
):
|
||||
"""A non-existent switch entity surfaces a ``switch_not_found`` error."""
|
||||
flow_id = await _start_user_flow(hass)
|
||||
user_input = make_entry_data(switch_entity_id="switch.ghost")
|
||||
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
# The error key is the field name; the exact key depends on whether the
|
||||
# entity selector also rejects the value. Either path must surface an error.
|
||||
assert any(k == CONF_SWITCH_ENTITY_ID for k in result["errors"])
|
||||
|
||||
|
||||
async def test_blank_name_shows_error(hass: HomeAssistant, gate_switch: str):
|
||||
"""An empty/whitespace-only name is rejected."""
|
||||
flow_id = await _start_user_flow(hass)
|
||||
user_input = make_entry_data(switch_entity_id=gate_switch, name=" ")
|
||||
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_NAME: "invalid_name"}
|
||||
|
||||
|
||||
async def test_notify_with_targets_succeeds(
|
||||
hass: HomeAssistant, gate_switch: str, notify_entity: str
|
||||
):
|
||||
"""Enabling notifications with at least one target is accepted."""
|
||||
flow_id = await _start_user_flow(hass)
|
||||
user_input = make_entry_data(
|
||||
switch_entity_id=gate_switch,
|
||||
name="Main Gate",
|
||||
notify_enabled=True,
|
||||
notify_entities=[notify_entity],
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_NOTIFY_ENABLED] is True
|
||||
assert result["data"][CONF_NOTIFY_ENTITIES] == [notify_entity]
|
||||
@@ -0,0 +1,188 @@
|
||||
"""Tests covering error paths in the controller."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.main_gate_controller.const import (
|
||||
DOMAIN,
|
||||
GATE_STATE_CLOSED,
|
||||
NOTIFY_DOMAIN,
|
||||
SERVICE_SEND_MESSAGE,
|
||||
)
|
||||
from tests.conftest import ServiceCalls, make_entry_data
|
||||
|
||||
|
||||
def _make_failing_handler(fail_on_domain: str, fail_on_service: str, message: str):
|
||||
async def _handler(call) -> None: # type: ignore[no-untyped-def]
|
||||
if call.domain == fail_on_domain and call.service == fail_on_service:
|
||||
raise HomeAssistantError(message)
|
||||
# Recording handled elsewhere – this helper is only for failure simulation.
|
||||
|
||||
return _handler
|
||||
|
||||
|
||||
async def test_notification_failure_continues_cycle(
|
||||
hass: HomeAssistant, service_calls: ServiceCalls
|
||||
):
|
||||
"""If a notify.send_message call fails, the cycle still finishes normally."""
|
||||
switch = "switch.main_gate_relay"
|
||||
hass.states.async_set(switch, "off")
|
||||
hass.states.async_set("notify.mobile_app_phone", "unknown")
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(
|
||||
switch_entity_id=switch,
|
||||
name="Main Gate",
|
||||
open_duration=1,
|
||||
pulse_duration=0.1,
|
||||
notify_enabled=True,
|
||||
notify_entities=["notify.mobile_app_phone"],
|
||||
),
|
||||
unique_id=switch,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
# Replace the recording notify handler with one that fails.
|
||||
hass.services.async_remove(NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE)
|
||||
hass.services.async_register(
|
||||
NOTIFY_DOMAIN,
|
||||
SERVICE_SEND_MESSAGE,
|
||||
_make_failing_handler(
|
||||
NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE, "deliberate failure"
|
||||
),
|
||||
)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
controller = entry.runtime_data
|
||||
assert controller.async_trigger() is True
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
assert controller.status == GATE_STATE_CLOSED
|
||||
assert controller.is_running is False
|
||||
|
||||
|
||||
async def test_notification_log_warning_emitted(
|
||||
hass: HomeAssistant, service_calls: ServiceCalls, caplog
|
||||
):
|
||||
"""The notify failure path emits a warning via the module logger."""
|
||||
switch = "switch.main_gate_relay"
|
||||
hass.states.async_set(switch, "off")
|
||||
hass.states.async_set("notify.mobile_app_phone", "unknown")
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(
|
||||
switch_entity_id=switch,
|
||||
name="Main Gate",
|
||||
open_duration=0.5,
|
||||
pulse_duration=0.1,
|
||||
notify_enabled=True,
|
||||
notify_entities=["notify.mobile_app_phone"],
|
||||
),
|
||||
unique_id=switch,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
hass.services.async_remove(NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE)
|
||||
|
||||
async def _failing_notify(call) -> None:
|
||||
raise HomeAssistantError("boom")
|
||||
|
||||
hass.services.async_register(NOTIFY_DOMAIN, SERVICE_SEND_MESSAGE, _failing_notify)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
controller = entry.runtime_data
|
||||
with caplog.at_level(logging.WARNING):
|
||||
controller.async_trigger()
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
assert any(
|
||||
"notification" in r.message.lower() and "boom" in r.message
|
||||
for r in caplog.records
|
||||
)
|
||||
|
||||
|
||||
async def test_switch_failure_aborts_cycle(
|
||||
hass: HomeAssistant, service_calls: ServiceCalls, caplog
|
||||
):
|
||||
"""A switch.turn_on failure aborts the cycle and resets state to closed."""
|
||||
switch = "switch.main_gate_relay"
|
||||
hass.states.async_set(switch, "off")
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(
|
||||
switch_entity_id=switch,
|
||||
name="Main Gate",
|
||||
open_duration=20,
|
||||
pulse_duration=0.1,
|
||||
notify_enabled=False,
|
||||
notify_entities=[],
|
||||
),
|
||||
unique_id=switch,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
# Replace the recording handler for switch.turn_off with a failing one.
|
||||
hass.services.async_remove("switch", "turn_off")
|
||||
|
||||
async def _failing_turn_off(call) -> None:
|
||||
raise HomeAssistantError("relay offline")
|
||||
|
||||
hass.services.async_register("switch", "turn_off", _failing_turn_off)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
controller = entry.runtime_data
|
||||
with caplog.at_level(logging.ERROR):
|
||||
controller.async_trigger()
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
assert controller.status == GATE_STATE_CLOSED
|
||||
assert controller.is_running is False
|
||||
assert any(
|
||||
"aborted" in r.message.lower() or "cycle aborted" in r.message.lower()
|
||||
for r in caplog.records
|
||||
)
|
||||
|
||||
|
||||
async def test_unload_cancels_running_cycle(
|
||||
hass: HomeAssistant, service_calls: ServiceCalls
|
||||
):
|
||||
"""Unloading an entry while a cycle runs cancels the background task."""
|
||||
switch = "switch.main_gate_relay"
|
||||
hass.states.async_set(switch, "off")
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(
|
||||
switch_entity_id=switch,
|
||||
name="Main Gate",
|
||||
open_duration=3,
|
||||
pulse_duration=0.2,
|
||||
notify_enabled=False,
|
||||
notify_entities=[],
|
||||
),
|
||||
unique_id=switch,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
controller = entry.runtime_data
|
||||
controller.async_trigger()
|
||||
# Immediately start unloading.
|
||||
assert await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert controller.is_running is False
|
||||
assert controller.status == GATE_STATE_CLOSED
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Tests for entity creation and device info."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from custom_components.main_gate_controller.const import (
|
||||
DOMAIN,
|
||||
GATE_STATE_CLOSED,
|
||||
MANUFACTURER,
|
||||
MODEL,
|
||||
)
|
||||
|
||||
|
||||
async def test_setup_creates_entities_and_device(
|
||||
hass: HomeAssistant, loaded_entry
|
||||
):
|
||||
"""A loaded entry creates one button, one sensor and one device."""
|
||||
entity_registry = er.async_get(hass)
|
||||
device_registry = dr.async_get(hass)
|
||||
|
||||
button_unique_id = f"{loaded_entry.entry_id}_open"
|
||||
sensor_unique_id = f"{loaded_entry.entry_id}_status"
|
||||
|
||||
button_entity_id = entity_registry.async_get_entity_id(
|
||||
Platform.BUTTON, DOMAIN, button_unique_id
|
||||
)
|
||||
sensor_entity_id = entity_registry.async_get_entity_id(
|
||||
Platform.SENSOR, DOMAIN, sensor_unique_id
|
||||
)
|
||||
assert button_entity_id is not None
|
||||
assert sensor_entity_id is not None
|
||||
|
||||
button_state = hass.states.get(button_entity_id)
|
||||
sensor_state = hass.states.get(sensor_entity_id)
|
||||
assert button_state is not None
|
||||
assert sensor_state is not None
|
||||
assert sensor_state.state == GATE_STATE_CLOSED
|
||||
|
||||
attrs = sensor_state.attributes
|
||||
assert attrs["remaining_seconds"] is None
|
||||
assert attrs["running"] is False
|
||||
assert attrs["duration"] == 20.0
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, loaded_entry.entry_id)})
|
||||
assert device is not None
|
||||
assert device.name == "Main Gate"
|
||||
assert device.manufacturer == MANUFACTURER
|
||||
assert device.model == MODEL
|
||||
|
||||
|
||||
async def test_setup_states_remain_consistent_after_restart(hass: HomeAssistant):
|
||||
"""After a (logical) restart the state is closed and not running."""
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from tests.conftest import make_entry_data
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(switch_entity_id="switch.main_gate_relay", name="Main Gate"),
|
||||
unique_id="switch.main_gate_relay",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
sensor_state = hass.states.get(f"sensor.{entry.title.lower().replace(' ', '_')}_status")
|
||||
assert sensor_state is None or sensor_state.state == GATE_STATE_CLOSED
|
||||
|
||||
|
||||
async def test_unload_entry(hass: HomeAssistant, loaded_entry):
|
||||
"""Unloading removes entities and clears the runtime data attribute."""
|
||||
# Capture the runtime data reference before unload – HA deletes the
|
||||
# attribute on unload by design (see homeassistant.config_entries).
|
||||
controller_before = loaded_entry.runtime_data
|
||||
assert controller_before is not None
|
||||
|
||||
assert await hass.config_entries.async_unload(loaded_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# The cycle was cancelled; the controller is reset to a safe closed state.
|
||||
assert controller_before.is_running is False
|
||||
assert controller_before.status == GATE_STATE_CLOSED
|
||||
assert not hasattr(loaded_entry, "runtime_data")
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Tests for the options flow and live settings updates."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from custom_components.main_gate_controller.const import (
|
||||
ATTR_DURATION,
|
||||
CONF_OPEN_DURATION,
|
||||
CONF_PULSE_DURATION,
|
||||
DOMAIN,
|
||||
)
|
||||
|
||||
|
||||
async def test_options_flow_updates_settings(
|
||||
hass: HomeAssistant, loaded_entry
|
||||
):
|
||||
"""The options flow updates the open duration and the sensor reflects it."""
|
||||
# Trigger an initial state refresh so the entity is registered.
|
||||
loaded_entry.runtime_data.async_publish_update()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
result = await hass.config_entries.options.async_init(loaded_entry.entry_id)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
new_input = {
|
||||
CONF_OPEN_DURATION: 5,
|
||||
CONF_PULSE_DURATION: 0.5,
|
||||
"notify_enabled": False,
|
||||
"notify_entities": [],
|
||||
"notify_opening_text": "Opening",
|
||||
"notify_closing_text": "Closing",
|
||||
}
|
||||
result = await hass.config_entries.options.async_configure(flow_id, new_input)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert loaded_entry.options[CONF_OPEN_DURATION] == 5
|
||||
assert loaded_entry.options[CONF_PULSE_DURATION] == 0.5
|
||||
|
||||
# Allow the update listener to push a refresh.
|
||||
await asyncio.sleep(0)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
controller = loaded_entry.runtime_data
|
||||
assert controller.open_duration == 5
|
||||
assert controller.pulse_duration == 0.5
|
||||
# The sensor attribute reflects the new duration as well.
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
registry = er.async_get(hass)
|
||||
sensor_id = registry.async_get_entity_id(
|
||||
"sensor", DOMAIN, f"{loaded_entry.entry_id}_status"
|
||||
)
|
||||
assert sensor_id is not None
|
||||
state = hass.states.get(sensor_id)
|
||||
assert state.attributes[ATTR_DURATION] == 5
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Tests for the countdown phase of the cycle."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.main_gate_controller.const import (
|
||||
ATTR_DURATION,
|
||||
ATTR_REMAINING_SECONDS,
|
||||
ATTR_RUNNING,
|
||||
DOMAIN,
|
||||
GATE_STATE_OPEN,
|
||||
)
|
||||
from tests.conftest import make_entry_data
|
||||
|
||||
|
||||
async def test_countdown_attributes_decrease(hass: HomeAssistant, service_calls):
|
||||
"""remaining_seconds decreases over the open phase and finishes at zero."""
|
||||
switch = "switch.fast_gate_relay"
|
||||
hass.states.async_set(switch, "off")
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Main Gate",
|
||||
data=make_entry_data(
|
||||
switch_entity_id=switch,
|
||||
name="Main Gate",
|
||||
open_duration=2,
|
||||
pulse_duration=0.1,
|
||||
notify_enabled=False,
|
||||
notify_entities=[],
|
||||
),
|
||||
unique_id=switch,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
controller = entry.runtime_data
|
||||
|
||||
# Manually step into the open phase and capture attribute changes.
|
||||
assert controller.async_trigger() is True
|
||||
|
||||
# Wait until the cycle is in the ``open`` state.
|
||||
while controller.status != GATE_STATE_OPEN:
|
||||
await asyncio.sleep(0.05)
|
||||
snapshot1 = controller.extra_state_attributes.copy()
|
||||
assert snapshot1[ATTR_DURATION] == 2
|
||||
assert snapshot1[ATTR_RUNNING] is True
|
||||
first_remaining = snapshot1[ATTR_REMAINING_SECONDS]
|
||||
assert first_remaining is not None and first_remaining >= 1
|
||||
|
||||
# Let the countdown progress for ~1 second before re-checking.
|
||||
await asyncio.sleep(1.0)
|
||||
snapshot2 = controller.extra_state_attributes.copy()
|
||||
assert snapshot2[ATTR_REMAINING_SECONDS] < first_remaining
|
||||
assert snapshot2[ATTR_RUNNING] is True
|
||||
|
||||
# Wait for completion.
|
||||
await controller.async_wait_until_idle()
|
||||
|
||||
after = controller.extra_state_attributes
|
||||
assert after[ATTR_RUNNING] is False
|
||||
assert after[ATTR_REMAINING_SECONDS] is None
|
||||
|
||||
|
||||
async def test_duration_reflects_configured_value(
|
||||
hass: HomeAssistant, loaded_entry
|
||||
):
|
||||
"""The ``duration`` attribute follows the configuration value."""
|
||||
controller = loaded_entry.runtime_data
|
||||
attrs = controller.extra_state_attributes
|
||||
assert attrs[ATTR_DURATION] == 20
|
||||
Reference in New Issue
Block a user