285 lines
9.2 KiB
Python
285 lines
9.2 KiB
Python
"""Tests for MB-Secure devices and entities."""
|
|
|
|
from copy import deepcopy
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import device_registry as dr
|
|
from homeassistant.helpers import entity_registry as er
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.mb_secure.const import CONF_TOKEN, DOMAIN
|
|
from custom_components.mb_secure.coordinator import MBSecureCoordinator
|
|
from custom_components.mb_secure.models import BridgeInfo, BridgeSnapshot
|
|
|
|
ENTRY_DATA = {
|
|
"host": "bridge.local",
|
|
"port": 8099,
|
|
CONF_TOKEN: "x" * 32,
|
|
}
|
|
|
|
|
|
def snapshot_payload() -> dict[str, object]:
|
|
"""Return a capability-complete normalized test snapshot."""
|
|
return {
|
|
"revision": 1,
|
|
"controllers": [
|
|
{
|
|
"id": "controller-1",
|
|
"name": "Main Controller",
|
|
"state": "online",
|
|
"capabilities": ["snapshot"],
|
|
}
|
|
],
|
|
"areas": [
|
|
{
|
|
"id": "area-1",
|
|
"controller_id": "controller-1",
|
|
"name": "Ground Floor",
|
|
"state": "disarmed",
|
|
"capabilities": ["arm_away", "disarm"],
|
|
}
|
|
],
|
|
"points": [
|
|
{
|
|
"id": "point-1",
|
|
"controller_id": "controller-1",
|
|
"area_id": "area-1",
|
|
"module_id": None,
|
|
"name": "Front Door",
|
|
"state": "closed",
|
|
"capabilities": ["binary_state"],
|
|
},
|
|
{
|
|
"id": "point-without-capability",
|
|
"controller_id": "controller-1",
|
|
"area_id": "area-1",
|
|
"module_id": None,
|
|
"name": "Unsupported Point",
|
|
"state": "unknown",
|
|
"capabilities": [],
|
|
},
|
|
],
|
|
"modules": [
|
|
{
|
|
"id": "module-1",
|
|
"controller_id": "controller-1",
|
|
"name": "Expansion Module",
|
|
"state": "online",
|
|
"capabilities": ["status"],
|
|
}
|
|
],
|
|
"outputs": [
|
|
{
|
|
"id": "output-1",
|
|
"controller_id": "controller-1",
|
|
"module_id": "module-1",
|
|
"name": "Indicator",
|
|
"state": "off",
|
|
"capabilities": ["set"],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
async def setup_entry(
|
|
hass: HomeAssistant,
|
|
*,
|
|
connected: bool = True,
|
|
payload: dict[str, object] | None = None,
|
|
) -> MockConfigEntry:
|
|
"""Set up an entry against mocked neutral Bridge API responses."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data=ENTRY_DATA,
|
|
unique_id="bridge-installation-1",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
info = BridgeInfo(
|
|
bridge_id="bridge-installation-1",
|
|
bridge_version="0.1.0",
|
|
api_version=1,
|
|
connected=connected,
|
|
capabilities=frozenset({"snapshot", "events", "commands"}),
|
|
)
|
|
snapshot = BridgeSnapshot.from_dict(payload or snapshot_payload())
|
|
with (
|
|
patch(
|
|
"custom_components.mb_secure.api.BridgeClient.async_get_info",
|
|
return_value=info,
|
|
),
|
|
patch(
|
|
"custom_components.mb_secure.api.BridgeClient.async_get_snapshot",
|
|
return_value=snapshot,
|
|
),
|
|
patch.object(MBSecureCoordinator, "async_start_event_listener"),
|
|
):
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
return entry
|
|
|
|
|
|
async def test_setup_creates_capability_gated_entities_and_hierarchy(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Only reported capabilities create stable entities and linked devices."""
|
|
entry = await setup_entry(hass)
|
|
registry = er.async_get(hass)
|
|
unique_ids = {
|
|
entity.unique_id
|
|
for entity in er.async_entries_for_config_entry(registry, entry.entry_id)
|
|
}
|
|
assert unique_ids == {
|
|
"bridge-installation-1:areas:area-1",
|
|
"bridge-installation-1:controllers:controller-1",
|
|
"bridge-installation-1:modules:module-1",
|
|
"bridge-installation-1:outputs:output-1",
|
|
"bridge-installation-1:points:point-1",
|
|
}
|
|
|
|
device_registry = dr.async_get(hass)
|
|
controller = device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "bridge-installation-1:controller:controller-1"),
|
|
entry.entry_id,
|
|
)
|
|
area = device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "bridge-installation-1:area:area-1"),
|
|
entry.entry_id,
|
|
)
|
|
module = device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "bridge-installation-1:module:module-1"),
|
|
entry.entry_id,
|
|
)
|
|
assert controller is not None
|
|
assert area is not None and area.via_device_id == controller.id
|
|
assert module is not None and module.via_device_id == controller.id
|
|
|
|
|
|
async def test_snapshot_topology_changes_add_and_remove_entities(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Snapshot reconciliation updates the runtime entity topology."""
|
|
entry = await setup_entry(hass)
|
|
coordinator = entry.runtime_data
|
|
registry = er.async_get(hass)
|
|
old_entry = registry.async_get_entity_id(
|
|
"binary_sensor",
|
|
DOMAIN,
|
|
"bridge-installation-1:points:point-1",
|
|
)
|
|
assert old_entry is not None
|
|
payload = deepcopy(snapshot_payload())
|
|
payload["revision"] = 2
|
|
points = payload["points"]
|
|
assert isinstance(points, list)
|
|
points.pop(0)
|
|
points.append(
|
|
{
|
|
"id": "point-2",
|
|
"controller_id": "controller-1",
|
|
"area_id": "area-1",
|
|
"module_id": None,
|
|
"name": "Back Door",
|
|
"state": "open",
|
|
"capabilities": ["binary_state"],
|
|
}
|
|
)
|
|
|
|
coordinator.async_set_updated_data(BridgeSnapshot.from_dict(payload))
|
|
await hass.async_block_till_done()
|
|
|
|
unique_ids = {
|
|
entity.unique_id
|
|
for entity in er.async_entries_for_config_entry(registry, entry.entry_id)
|
|
}
|
|
assert "bridge-installation-1:points:point-1" not in unique_ids
|
|
assert "bridge-installation-1:points:point-2" in unique_ids
|
|
assert hass.states.get(old_entry) is None
|
|
|
|
|
|
async def test_panel_connection_loss_marks_all_entities_unavailable(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""A reachable bridge does not hide loss of its panel connection."""
|
|
entry = await setup_entry(hass, connected=False)
|
|
registry = er.async_get(hass)
|
|
|
|
for entity in er.async_entries_for_config_entry(registry, entry.entry_id):
|
|
state = hass.states.get(entity.entity_id)
|
|
assert state is not None
|
|
assert state.state == "unavailable"
|
|
|
|
|
|
async def test_unknown_binary_states_remain_unknown(hass: HomeAssistant) -> None:
|
|
"""Unknown neutral states are never converted into a false clear state."""
|
|
payload = snapshot_payload()
|
|
points = payload["points"]
|
|
outputs = payload["outputs"]
|
|
assert isinstance(points, list) and isinstance(outputs, list)
|
|
points[0]["state"] = "future_state"
|
|
outputs[0]["state"] = "future_state"
|
|
entry = await setup_entry(hass, payload=payload)
|
|
registry = er.async_get(hass)
|
|
|
|
for unique_id in (
|
|
"bridge-installation-1:points:point-1",
|
|
"bridge-installation-1:outputs:output-1",
|
|
):
|
|
entity_id = next(
|
|
entity.entity_id
|
|
for entity in er.async_entries_for_config_entry(registry, entry.entry_id)
|
|
if entity.unique_id == unique_id
|
|
)
|
|
state = hass.states.get(entity_id)
|
|
assert state is not None
|
|
assert state.state == "unknown"
|
|
|
|
|
|
async def test_entity_services_use_scoped_bridge_commands(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Entity services send only normalized commands for their own object IDs."""
|
|
entry = await setup_entry(hass)
|
|
coordinator = entry.runtime_data
|
|
coordinator.client.async_arm_area = AsyncMock()
|
|
coordinator.client.async_disarm_area = AsyncMock()
|
|
coordinator.client.async_set_output = AsyncMock()
|
|
coordinator.async_request_refresh = AsyncMock()
|
|
registry = er.async_get(hass)
|
|
|
|
area_entity_id = registry.async_get_entity_id(
|
|
"alarm_control_panel",
|
|
DOMAIN,
|
|
"bridge-installation-1:areas:area-1",
|
|
)
|
|
output_entity_id = registry.async_get_entity_id(
|
|
"switch",
|
|
DOMAIN,
|
|
"bridge-installation-1:outputs:output-1",
|
|
)
|
|
assert area_entity_id is not None and output_entity_id is not None
|
|
|
|
await hass.services.async_call(
|
|
"alarm_control_panel",
|
|
"alarm_arm_away",
|
|
{"entity_id": area_entity_id},
|
|
blocking=True,
|
|
)
|
|
await hass.services.async_call(
|
|
"alarm_control_panel",
|
|
"alarm_disarm",
|
|
{"entity_id": area_entity_id},
|
|
blocking=True,
|
|
)
|
|
await hass.services.async_call(
|
|
"switch",
|
|
"turn_on",
|
|
{"entity_id": output_entity_id},
|
|
blocking=True,
|
|
)
|
|
|
|
coordinator.client.async_arm_area.assert_awaited_once_with("area-1", "away")
|
|
coordinator.client.async_disarm_area.assert_awaited_once_with("area-1")
|
|
coordinator.client.async_set_output.assert_awaited_once_with("output-1", "on")
|
|
assert coordinator.async_request_refresh.await_count == 3
|