mirror of
https://github.com/Matysh/houseplan-card
synced 2026-09-29 03:09:36 +00:00
Matysh
@@ -251,3 +251,127 @@ async def test_teardown_releases_frames_and_subscriptions(
|
||||
assert coordinator.closed is True
|
||||
assert coordinator.frames_for_space("floor") == []
|
||||
assert coordinator._unsub_sources == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_presence_only_profile_reports_without_geometry(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
document = _stored("presence_v1")
|
||||
radar = document["config"]["markers"][0]["radar"]
|
||||
radar["sources"] = {"occupancy_entity": "binary_sensor.radar_presence"}
|
||||
hass.states.async_set("binary_sensor.radar_presence", "on")
|
||||
|
||||
coordinator = await _coordinator(hass, document)
|
||||
frame = coordinator.frame("radar")
|
||||
assert frame is not None
|
||||
assert frame["health"] == "ok"
|
||||
assert frame["reported_presence"] is True
|
||||
assert frame["complete"] is True
|
||||
assert frame["expires_at"] is None
|
||||
assert coordinator.inspect("radar")["capabilities"] == ["reported_presence"]
|
||||
coordinator.teardown()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zone_profile_normalizes_occupancy_and_count(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
document = _stored("zones_v1")
|
||||
radar = document["config"]["markers"][0]["radar"]
|
||||
radar["sources"] = {"zones": [
|
||||
{"id": "desk", "entity_id": "binary_sensor.desk", "kind": "occupancy"},
|
||||
{"id": "people", "entity_id": "sensor.people", "kind": "count"},
|
||||
]}
|
||||
hass.states.async_set("binary_sensor.desk", "on")
|
||||
hass.states.async_set("sensor.people", "2")
|
||||
|
||||
coordinator = await _coordinator(hass, document)
|
||||
frame = coordinator.frame("radar")
|
||||
assert frame is not None
|
||||
assert frame["health"] == "ok"
|
||||
assert frame["complete"] is True
|
||||
assert frame["zones"] == [
|
||||
{"id": "desk", "state": True}, {"id": "people", "state": 2.0},
|
||||
]
|
||||
assert coordinator.inspect("radar")["capabilities"] == ["zone_state"]
|
||||
coordinator.teardown()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_range_profile_clips_arc_to_owner_room(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
document = _stored("range_v1")
|
||||
radar = document["config"]["markers"][0]["radar"]
|
||||
radar["sources"] = {"ranges": [{
|
||||
"id": "distance", "entity_id": "sensor.radar_distance", "unit": "m",
|
||||
"presence_entity": "binary_sensor.radar_presence",
|
||||
}]}
|
||||
hass.states.async_set("binary_sensor.radar_presence", "on")
|
||||
hass.states.async_set("sensor.radar_distance", "2")
|
||||
|
||||
coordinator = await _coordinator(hass, document)
|
||||
frame = coordinator.frame("radar")
|
||||
assert frame is not None
|
||||
assert frame["health"] == "ok"
|
||||
assert frame["complete"] is True
|
||||
assert frame["ranges"][0]["radius"] == pytest.approx(200 / 1200)
|
||||
assert frame["ranges"][0]["segments"]
|
||||
assert coordinator.inspect("radar")["capabilities"] == ["range"]
|
||||
coordinator.teardown()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_polar_profile_projects_explicit_bearing(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
document = _stored("polar_v1")
|
||||
radar = document["config"]["markers"][0]["radar"]
|
||||
radar["sources"] = {"slots": [{
|
||||
"id": "target_1", "distance_entity": "sensor.radar_distance",
|
||||
"angle_entity": "sensor.radar_angle", "unit": "cm",
|
||||
"angle_unit": "degrees", "angle_zero": "forward",
|
||||
"angle_clockwise": True,
|
||||
}]}
|
||||
hass.states.async_set("sensor.radar_distance", "100")
|
||||
hass.states.async_set("sensor.radar_angle", "90")
|
||||
|
||||
coordinator = await _coordinator(hass, document)
|
||||
frame = coordinator.frame("radar")
|
||||
assert frame is not None
|
||||
assert frame["health"] == "ok"
|
||||
assert frame["targets"][0]["x"] == pytest.approx(.5 + 100 / 1200)
|
||||
assert frame["targets"][0]["y"] == pytest.approx(.5)
|
||||
coordinator.teardown()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_coordinator_query_and_listener_lifecycle(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
hass.states.async_set("binary_sensor.radar_presence", "off")
|
||||
coordinator = await _coordinator(hass, _stored())
|
||||
public_events = []
|
||||
internal_events = []
|
||||
remove_public = coordinator.add_listener(
|
||||
lambda marker_id, frame: public_events.append((marker_id, frame)),
|
||||
)
|
||||
remove_internal = coordinator.add_internal_listener(
|
||||
lambda marker_id, frame: internal_events.append((marker_id, frame)),
|
||||
)
|
||||
assert public_events[0][0] == "radar"
|
||||
assert coordinator.has_space("floor") is True
|
||||
assert coordinator.has_space("missing") is False
|
||||
assert coordinator.space_for_marker("missing") is None
|
||||
assert coordinator.marker_config("radar")["id"] == "radar"
|
||||
assert coordinator.marker_config("missing") is None
|
||||
assert coordinator.source_ids("missing") == set()
|
||||
assert coordinator.inspect("missing")["health"] == "not_configured"
|
||||
|
||||
coordinator._publish_all(force=True)
|
||||
assert internal_events[-1][0] == "radar"
|
||||
remove_public()
|
||||
remove_internal()
|
||||
assert coordinator._unsub_tick is None
|
||||
coordinator.teardown()
|
||||
|
||||
@@ -0,0 +1,317 @@
|
||||
"""Permission, lifecycle and back-pressure witnesses for radar WebSockets."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from custom_components.houseplan import radar_websocket as radar_ws
|
||||
from custom_components.houseplan.radar import RadarCoordinator
|
||||
|
||||
|
||||
class _Permissions:
|
||||
def __init__(self, allowed: bool = True) -> None:
|
||||
self.allowed = allowed
|
||||
|
||||
def check_entity(self, _entity_id: str, _policy: str) -> bool:
|
||||
return self.allowed
|
||||
|
||||
|
||||
class _Connection:
|
||||
def __init__(self, *, allowed: bool = True) -> None:
|
||||
self.user = SimpleNamespace(id="user-1", permissions=_Permissions(allowed))
|
||||
self.subscriptions: dict[int, object] = {}
|
||||
self.results: list[tuple[int, object]] = []
|
||||
self.errors: list[tuple[int, str, str]] = []
|
||||
self.events: list[tuple[int, object]] = []
|
||||
|
||||
def send_result(self, msg_id: int, result: object) -> None:
|
||||
self.results.append((msg_id, result))
|
||||
|
||||
def send_error(self, msg_id: int, code: str, message: str) -> None:
|
||||
self.errors.append((msg_id, code, message))
|
||||
|
||||
def send_event(self, msg_id: int, event: object) -> None:
|
||||
self.events.append((msg_id, event))
|
||||
|
||||
|
||||
class _Coordinator:
|
||||
def __init__(self) -> None:
|
||||
self.closed = False
|
||||
self.server_session_id = "session-1"
|
||||
self.config_rev = 7
|
||||
self.config = {"markers": []}
|
||||
self.radars = {"radar": {"id": "radar", "radar": {"version": 1}}}
|
||||
self.listener = None
|
||||
self.listener_removed = False
|
||||
self.external_cleanup = None
|
||||
|
||||
def has_space(self, space_id: str) -> bool:
|
||||
return space_id == "floor"
|
||||
|
||||
def source_ids(self, marker_id: str) -> set[str]:
|
||||
return {"sensor.x"} if marker_id == "radar" else set()
|
||||
|
||||
def space_for_marker(self, marker_id: str) -> str | None:
|
||||
return "floor" if marker_id == "radar" else "other"
|
||||
|
||||
def frames_for_space(self, _space_id: str) -> list[dict]:
|
||||
return [{"marker_id": "radar", "seq": 3, "targets": []}]
|
||||
|
||||
@staticmethod
|
||||
def public_frame(frame: dict) -> dict:
|
||||
return {**frame, "public": True}
|
||||
|
||||
def add_listener(self, listener, *, publish_initial: bool = True):
|
||||
assert publish_initial is False
|
||||
self.listener = listener
|
||||
|
||||
def remove() -> None:
|
||||
self.listener_removed = True
|
||||
|
||||
return remove
|
||||
|
||||
def inspect(self, marker_id: str) -> dict:
|
||||
return {"marker_id": marker_id, "health": "ok"}
|
||||
|
||||
def inspect_draft(self, marker: dict, radar: dict) -> dict:
|
||||
return {"marker_id": marker["id"], "profile": radar.get("profile")}
|
||||
|
||||
def marker_config(self, marker_id: str) -> dict | None:
|
||||
return self.radars.get(marker_id)
|
||||
|
||||
def add_external_cleanup(self, cleanup):
|
||||
self.external_cleanup = cleanup
|
||||
|
||||
def unregister() -> None:
|
||||
self.external_cleanup = None
|
||||
|
||||
return unregister
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_radar_ws_state():
|
||||
radar_ws._INSPECT_CALLS.clear()
|
||||
radar_ws._ACTIVE_SUBSCRIPTIONS.clear()
|
||||
radar_ws._ACTIVE_SETUP.clear()
|
||||
yield
|
||||
radar_ws._INSPECT_CALLS.clear()
|
||||
radar_ws._ACTIVE_SUBSCRIPTIONS.clear()
|
||||
radar_ws._ACTIVE_SETUP.clear()
|
||||
|
||||
|
||||
def test_registers_all_radar_commands(monkeypatch, hass) -> None:
|
||||
registered = []
|
||||
monkeypatch.setattr(radar_ws.websocket_api, "async_register_command",
|
||||
lambda _hass, command: registered.append(command))
|
||||
radar_ws.async_register(hass)
|
||||
assert registered == [
|
||||
radar_ws.ws_radar_subscribe,
|
||||
radar_ws.ws_radar_setup_inspect,
|
||||
radar_ws.ws_radar_setup_subscribe,
|
||||
]
|
||||
|
||||
|
||||
def test_coordinator_lookup_and_read_permissions(monkeypatch, hass) -> None:
|
||||
connection = _Connection()
|
||||
monkeypatch.setattr(radar_ws, "get_data", lambda _hass: None)
|
||||
assert radar_ws._coordinator(hass, connection, 1) is None
|
||||
assert connection.errors[-1][1] == "not_ready"
|
||||
|
||||
coordinator = RadarCoordinator(hass, SimpleNamespace())
|
||||
monkeypatch.setattr(
|
||||
radar_ws, "get_data", lambda _hass: SimpleNamespace(radar_coordinator=coordinator),
|
||||
)
|
||||
assert radar_ws._coordinator(hass, connection, 2) is coordinator
|
||||
coordinator.closed = True
|
||||
assert radar_ws._coordinator(hass, connection, 3) is None
|
||||
assert radar_ws._can_read(connection, {"sensor.x"}) is True
|
||||
connection.user.permissions.allowed = False
|
||||
assert radar_ws._can_read(connection, {"sensor.x"}) is False
|
||||
connection.user.permissions = None
|
||||
assert radar_ws._can_read(connection, set()) is False
|
||||
|
||||
|
||||
def test_rate_limit_prunes_old_calls() -> None:
|
||||
now = radar_ws.time.monotonic()
|
||||
radar_ws._INSPECT_CALLS["user"].extend([now] * 10)
|
||||
assert radar_ws._rate_limit("user") is False
|
||||
radar_ws._INSPECT_CALLS["user"].clear()
|
||||
radar_ws._INSPECT_CALLS["user"].append(now - 61)
|
||||
assert radar_ws._rate_limit("user") is True
|
||||
|
||||
|
||||
def test_live_subscribe_publishes_filters_and_cleans_up(monkeypatch, hass) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
monkeypatch.setattr(radar_ws, "_coordinator", lambda *_args: coordinator)
|
||||
radar_ws.ws_radar_subscribe(hass, connection, {"id": 1, "space_id": "floor"})
|
||||
|
||||
assert connection.results[-1][1] == {"server_session_id": "session-1", "config_rev": 7}
|
||||
assert connection.events[-1][1]["public"] is True
|
||||
assert radar_ws._ACTIVE_SUBSCRIPTIONS["user-1"] == 1
|
||||
assert coordinator.listener is not None
|
||||
coordinator.listener("other", {"seq": 4})
|
||||
assert len(connection.events) == 1
|
||||
connection.user.permissions.allowed = False
|
||||
coordinator.listener("radar", {"seq": 5})
|
||||
assert connection.events[-1][1]["health"] == "restricted"
|
||||
|
||||
connection.subscriptions[1]()
|
||||
assert coordinator.listener_removed is True
|
||||
assert "user-1" not in radar_ws._ACTIVE_SUBSCRIPTIONS
|
||||
|
||||
|
||||
def test_live_subscribe_rejects_unknown_space_and_connection_limit(monkeypatch, hass) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
monkeypatch.setattr(radar_ws, "_coordinator", lambda *_args: coordinator)
|
||||
radar_ws.ws_radar_subscribe(hass, connection, {"id": 1, "space_id": "missing"})
|
||||
assert connection.errors[-1][1] == "invalid_selection"
|
||||
radar_ws._ACTIVE_SUBSCRIPTIONS["user-1"] = 4
|
||||
radar_ws.ws_radar_subscribe(hass, connection, {"id": 2, "space_id": "floor"})
|
||||
assert connection.errors[-1][1] == "rate_limited"
|
||||
|
||||
|
||||
def test_draft_validation_size_permission_and_success(monkeypatch) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
marker = {"id": "radar"}
|
||||
radar = {"profile": "cartesian_v1"}
|
||||
monkeypatch.setattr(
|
||||
radar_ws, "validate_radar_draft", lambda *_args: (marker, {"sensor.x"}),
|
||||
)
|
||||
message = {"id": 1, "marker_id": "radar", "draft_sources": {"radar": radar}}
|
||||
assert radar_ws._validated_draft(coordinator, message, connection) == (
|
||||
marker, radar, {"sensor.x"},
|
||||
)
|
||||
connection.user.permissions.allowed = False
|
||||
assert radar_ws._validated_draft(coordinator, message, connection) is None
|
||||
assert connection.errors[-1][1] == "source_restricted"
|
||||
huge = {"id": 2, "marker_id": "radar", "draft_sources": {"x": "x" * 66000}}
|
||||
assert radar_ws._validated_draft(coordinator, huge, connection) is None
|
||||
assert connection.errors[-1][1] == "invalid_selection"
|
||||
|
||||
|
||||
def test_draft_validation_reports_stable_radar_error(monkeypatch) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
|
||||
def reject(*_args):
|
||||
raise radar_ws.RadarValidationError("bad")
|
||||
|
||||
monkeypatch.setattr(radar_ws, "validate_radar_draft", reject)
|
||||
assert radar_ws._validated_draft(
|
||||
coordinator,
|
||||
{"id": 1, "marker_id": "radar", "draft_sources": {"radar": {}}},
|
||||
connection,
|
||||
) is None
|
||||
assert connection.errors[-1][1] == "invalid_radar"
|
||||
|
||||
|
||||
def test_setup_inspect_permissions_limits_saved_and_draft(monkeypatch, hass) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
monkeypatch.setattr(radar_ws, "_coordinator", lambda *_args: coordinator)
|
||||
monkeypatch.setattr(radar_ws, "may_write", lambda *_args: False)
|
||||
radar_ws.ws_radar_setup_inspect(hass, connection, {"id": 1, "marker_id": "radar"})
|
||||
assert connection.errors[-1][1] == "unauthorized"
|
||||
|
||||
monkeypatch.setattr(radar_ws, "may_write", lambda *_args: True)
|
||||
connection.user.permissions.allowed = False
|
||||
radar_ws.ws_radar_setup_inspect(hass, connection, {"id": 2, "marker_id": "radar"})
|
||||
assert connection.errors[-1][1] == "source_restricted"
|
||||
connection.user.permissions.allowed = True
|
||||
radar_ws.ws_radar_setup_inspect(hass, connection, {"id": 3, "marker_id": "radar"})
|
||||
assert connection.results[-1][1]["health"] == "ok"
|
||||
|
||||
monkeypatch.setattr(
|
||||
radar_ws, "_validated_draft",
|
||||
lambda *_args: ({"id": "radar"}, {"profile": "range_v1"}, set()),
|
||||
)
|
||||
radar_ws.ws_radar_setup_inspect(
|
||||
hass, connection,
|
||||
{"id": 4, "marker_id": "radar", "draft_sources": {}},
|
||||
)
|
||||
assert connection.results[-1][1]["profile"] == "range_v1"
|
||||
monkeypatch.setattr(radar_ws, "_rate_limit", lambda _user_id: False)
|
||||
radar_ws.ws_radar_setup_inspect(hass, connection, {"id": 5, "marker_id": "radar"})
|
||||
assert connection.errors[-1][1] == "rate_limited"
|
||||
|
||||
|
||||
def test_setup_subscribe_guards_and_missing_marker(monkeypatch, hass) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
monkeypatch.setattr(radar_ws, "_coordinator", lambda *_args: coordinator)
|
||||
monkeypatch.setattr(radar_ws, "may_write", lambda *_args: False)
|
||||
base = {"marker_id": "radar", "expected_config_rev": 7}
|
||||
radar_ws.ws_radar_setup_subscribe(hass, connection, {"id": 1, **base})
|
||||
assert connection.errors[-1][1] == "unauthorized"
|
||||
|
||||
monkeypatch.setattr(radar_ws, "may_write", lambda *_args: True)
|
||||
radar_ws.ws_radar_setup_subscribe(
|
||||
hass, connection, {"id": 2, **base, "expected_config_rev": 6},
|
||||
)
|
||||
assert connection.errors[-1][1] == "conflict"
|
||||
radar_ws._ACTIVE_SETUP.add(("user-1", "radar"))
|
||||
radar_ws.ws_radar_setup_subscribe(hass, connection, {"id": 3, **base})
|
||||
assert connection.errors[-1][1] == "rate_limited"
|
||||
radar_ws._ACTIVE_SETUP.clear()
|
||||
coordinator.radars.clear()
|
||||
radar_ws.ws_radar_setup_subscribe(hass, connection, {"id": 4, **base})
|
||||
assert connection.errors[-1][1] == "invalid_selection"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_subscription_coalesces_restricts_and_removes(monkeypatch, hass) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection()
|
||||
callbacks = []
|
||||
cleanups = []
|
||||
monkeypatch.setattr(radar_ws, "_coordinator", lambda *_args: coordinator)
|
||||
monkeypatch.setattr(radar_ws, "may_write", lambda *_args: True)
|
||||
|
||||
def track(_hass, _ids, callback):
|
||||
callbacks.append(callback)
|
||||
|
||||
def cleanup() -> None:
|
||||
cleanups.append(True)
|
||||
|
||||
return cleanup
|
||||
|
||||
monkeypatch.setattr(radar_ws, "async_track_state_report_event", track)
|
||||
monkeypatch.setattr(radar_ws, "async_track_state_change_event", track)
|
||||
message = {"id": 9, "marker_id": "radar", "expected_config_rev": 7}
|
||||
radar_ws.ws_radar_setup_subscribe(hass, connection, message)
|
||||
assert connection.results
|
||||
assert connection.events[-1][1]["marker_id"] == "radar"
|
||||
assert ("user-1", "radar") in radar_ws._ACTIVE_SETUP
|
||||
|
||||
callbacks[0](None)
|
||||
callbacks[1](None)
|
||||
connection.user.permissions.allowed = False
|
||||
await asyncio.sleep(1 / radar_ws.MAX_FRAME_HZ + .05)
|
||||
assert connection.events[-1][1]["health"] == "restricted"
|
||||
|
||||
connection.subscriptions[9]()
|
||||
connection.subscriptions[9]()
|
||||
assert len(cleanups) == 2
|
||||
assert ("user-1", "radar") not in radar_ws._ACTIVE_SETUP
|
||||
assert coordinator.external_cleanup is None
|
||||
|
||||
|
||||
def test_setup_subscribe_draft_failure_and_source_restriction(monkeypatch, hass) -> None:
|
||||
coordinator = _Coordinator()
|
||||
connection = _Connection(allowed=False)
|
||||
monkeypatch.setattr(radar_ws, "_coordinator", lambda *_args: coordinator)
|
||||
monkeypatch.setattr(radar_ws, "may_write", lambda *_args: True)
|
||||
base = {"marker_id": "radar", "expected_config_rev": 7}
|
||||
radar_ws.ws_radar_setup_subscribe(hass, connection, {"id": 1, **base})
|
||||
assert connection.errors[-1][1] == "source_restricted"
|
||||
|
||||
monkeypatch.setattr(radar_ws, "_validated_draft", lambda *_args: None)
|
||||
radar_ws.ws_radar_setup_subscribe(
|
||||
hass, connection, {"id": 2, **base, "draft_sources": {}},
|
||||
)
|
||||
assert 2 not in connection.subscriptions
|
||||
@@ -6,7 +6,11 @@ import copy
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
|
||||
from custom_components.houseplan.radar_validation import validate_marker_radars
|
||||
from custom_components.houseplan.radar_validation import (
|
||||
radar_source_entity_ids,
|
||||
validate_marker_radars,
|
||||
validate_radar_draft,
|
||||
)
|
||||
|
||||
|
||||
def _config():
|
||||
@@ -105,3 +109,174 @@ def test_stage2_zone_polygon_and_stage3_cross_space_group_fail_closed():
|
||||
}]}
|
||||
with pytest.raises(vol.Invalid, match="zero area"):
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("profile", [
|
||||
"esphome_ld2450_v1", "cartesian_v1", "polar_v1", "range_v1",
|
||||
"zones_v1", "presence_v1",
|
||||
])
|
||||
def test_all_stage1_profiles_have_a_valid_explicit_source_contract(profile):
|
||||
config = _config()
|
||||
radar = config["markers"][0]["radar"]
|
||||
radar["profile"] = profile
|
||||
if profile == "esphome_ld2450_v1":
|
||||
radar["sources"]["slots"][0]["unit"] = "mm"
|
||||
elif profile == "polar_v1":
|
||||
radar["sources"] = {"slots": [{
|
||||
"id": "one", "distance_entity": "sensor.distance",
|
||||
"angle_entity": "sensor.angle", "unit": "cm",
|
||||
"angle_unit": "degrees", "angle_zero": "forward",
|
||||
"angle_clockwise": True,
|
||||
}]}
|
||||
elif profile == "range_v1":
|
||||
radar["sources"] = {"ranges": [{
|
||||
"id": "range", "entity_id": "sensor.distance", "unit": "m",
|
||||
"presence_entity": "binary_sensor.presence",
|
||||
}]}
|
||||
elif profile == "zones_v1":
|
||||
radar["sources"] = {"zones": [
|
||||
{"id": "occupied", "entity_id": "binary_sensor.zone", "kind": "occupancy"},
|
||||
{"id": "count", "entity_id": "sensor.zone_count", "kind": "count"},
|
||||
]}
|
||||
elif profile == "presence_v1":
|
||||
radar["sources"] = {"occupancy_entity": "binary_sensor.presence"}
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
|
||||
|
||||
def test_stage2_metadata_and_source_inventory_are_validated_together():
|
||||
config = _config()
|
||||
radar = config["markers"][0]["radar"]
|
||||
radar["sources"].update({
|
||||
"occupancy_entity": "binary_sensor.presence",
|
||||
"count_entity": "sensor.count",
|
||||
"availability_entity": "binary_sensor.available",
|
||||
})
|
||||
radar["sources"]["slots"][0]["presence_entity"] = "binary_sensor.slot"
|
||||
radar["zones"] = {
|
||||
"local": [{
|
||||
"id": "desk", "name": "Desk", "poly": [
|
||||
{"x": 0, "y": 0}, {"x": 100, "y": 0},
|
||||
{"x": 100, "y": 100}, {"x": 0, "y": 100},
|
||||
],
|
||||
"state": {"kind": "occupancy", "entity_id": "binary_sensor.desk"},
|
||||
}],
|
||||
"hardware": {
|
||||
"adapter": "esphome_ld2450_numbers_v1",
|
||||
"mode_entity": "select.zone_mode",
|
||||
"slots": [{
|
||||
"slot": index,
|
||||
"x1_entity": f"number.zone_{index}_x1",
|
||||
"y1_entity": f"number.zone_{index}_y1",
|
||||
"x2_entity": f"number.zone_{index}_x2",
|
||||
"y2_entity": f"number.zone_{index}_y2",
|
||||
} for index in range(1, 4)],
|
||||
},
|
||||
}
|
||||
radar["reflectors"] = [{
|
||||
"id": "mirror", "name": "Mirror", "enabled": True,
|
||||
"a": {"x": 0, "y": 0}, "b": {"x": 100, "y": 0},
|
||||
}]
|
||||
radar["allowed_room_ids"] = ["living"]
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
ids = radar_source_entity_ids(radar)
|
||||
assert {"sensor.x", "sensor.y", "binary_sensor.desk", "select.zone_mode",
|
||||
"number.zone_3_y2"} <= ids
|
||||
assert radar_source_entity_ids(None) == set()
|
||||
assert radar_source_entity_ids({"sources": []}) == set()
|
||||
|
||||
|
||||
def test_valid_fusion_and_room_output_settings():
|
||||
config = _config()
|
||||
second = copy.deepcopy(config["markers"][0])
|
||||
second["id"] = "radar-two"
|
||||
second["binding"] = "device:def"
|
||||
second["radar"]["mount"]["installation_id"] = "install-2"
|
||||
second["radar"]["sources"]["slots"][0].update(
|
||||
id="two", x_entity="sensor.x2", y_entity="sensor.y2",
|
||||
)
|
||||
config["markers"].append(second)
|
||||
config["settings"]["radar"] = {
|
||||
"version": 1, "show_live": False,
|
||||
"fusion_groups": [{
|
||||
"id": "living-pair", "enabled": True, "space_id": "floor",
|
||||
"marker_ids": ["radar", "radar-two"],
|
||||
}],
|
||||
"room_outputs": [{
|
||||
"id": "living-output", "space_id": "floor", "room_id": "living",
|
||||
"presence": True, "estimated_count": False,
|
||||
}],
|
||||
}
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mutation", [
|
||||
lambda r: r.update(enabled="yes"),
|
||||
lambda r: r.update(show_live="yes"),
|
||||
lambda r: r.update(profile="magic_v1"),
|
||||
lambda r: r.update(room_id="missing"),
|
||||
lambda r: r["mount"].update(installation_id="bad id"),
|
||||
lambda r: r["mount"].update(heading_deg=360),
|
||||
lambda r: r["mount"].update(range_cm=0),
|
||||
lambda r: r["mount"].update(fov_deg=361),
|
||||
lambda r: r["calibration"].update(method="guess"),
|
||||
lambda r: r["calibration"].update(mirror=1),
|
||||
lambda r: r["calibration"].update(refs=[{}]),
|
||||
lambda r: r["calibration"].update(rms_cm=31),
|
||||
lambda r: r.update(allowed_room_ids=["living", "living"]),
|
||||
])
|
||||
def test_invalid_core_mount_and_calibration_contracts_are_rejected(mutation):
|
||||
config = _config()
|
||||
mutation(config["markers"][0]["radar"])
|
||||
with pytest.raises(vol.Invalid):
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sources", [
|
||||
{},
|
||||
{"slots": [{"id": "one", "x_entity": "sensor.x", "y_entity": "sensor.y",
|
||||
"unit": "mm", "x_sign": 0}]},
|
||||
{"slots": [{"id": "one", "x_entity": "sensor.x", "y_entity": "sensor.y",
|
||||
"unit": "mm", "swap_xy": "yes"}]},
|
||||
{"slots": [{"id": "one", "x_entity": "sensor.x", "y_entity": "sensor.y",
|
||||
"unit": "mm", "presence_entity": "sensor.wrong"}]},
|
||||
])
|
||||
def test_invalid_cartesian_sources_are_rejected(sources):
|
||||
config = _config()
|
||||
config["markers"][0]["radar"]["sources"] = sources
|
||||
with pytest.raises(vol.Invalid):
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", [
|
||||
{"unit": "yards"}, {"angle_unit": "turns"}, {"angle_zero": "north"},
|
||||
{"angle_clockwise": 1}, {"angle_entity": "sensor.distance"},
|
||||
])
|
||||
def test_invalid_polar_sources_are_rejected(change):
|
||||
config = _config()
|
||||
radar = config["markers"][0]["radar"]
|
||||
radar["profile"] = "polar_v1"
|
||||
slot = {
|
||||
"id": "one", "distance_entity": "sensor.distance", "angle_entity": "sensor.angle",
|
||||
"unit": "cm", "angle_unit": "degrees", "angle_zero": "forward",
|
||||
"angle_clockwise": True,
|
||||
}
|
||||
slot.update(change)
|
||||
radar["sources"] = {"slots": [slot]}
|
||||
with pytest.raises(vol.Invalid):
|
||||
validate_marker_radars(config, validate_all=True)
|
||||
|
||||
|
||||
def test_draft_rejects_missing_removed_and_virtual_marker():
|
||||
config = _config()
|
||||
radar = config["markers"][0]["radar"]
|
||||
candidate, source_ids = validate_radar_draft(config, "radar", radar)
|
||||
assert candidate["id"] == "radar" and source_ids == {"sensor.x", "sensor.y"}
|
||||
for marker_id in ("missing",):
|
||||
with pytest.raises(vol.Invalid, match="setup marker"):
|
||||
validate_radar_draft(config, marker_id, radar)
|
||||
config["markers"][0]["removed"] = True
|
||||
with pytest.raises(vol.Invalid, match="setup marker"):
|
||||
validate_radar_draft(config, "radar", radar)
|
||||
config["markers"][0].update(removed=False, binding="virtual")
|
||||
with pytest.raises(vol.Invalid, match="setup marker"):
|
||||
validate_radar_draft(config, "radar", radar)
|
||||
|
||||
Reference in New Issue
Block a user