703 lines
25 KiB
Python
703 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from time import perf_counter
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.v1.actuators import _deduplicate_actuator_ids
|
|
from app.actuators.cache_db import DashboardCache
|
|
from app.actuators.lifecycle import ActuatorReconciliationService
|
|
from app.actuators.models import BehaviorPattern, JobStatus, ModelSnapshot
|
|
from app.actuators.store import ActuatorStore
|
|
from app.behavior.engine import BehaviorEngine
|
|
from app.config import Settings
|
|
from app.ha.discovery import DiscoveredEntity
|
|
from app.ha.discovery import discover_entities
|
|
from app.ha.history import (
|
|
EntityHistorySeries,
|
|
LogbookEntry,
|
|
NumericHistoryPoint,
|
|
StateHistorySeries,
|
|
)
|
|
from app.ha.models import HaAutomationSummary, HaEntitySummary
|
|
from app.ha.reader import HaReader
|
|
from app.main import app
|
|
from app.ml.registry.model_registry import ModelRegistry
|
|
|
|
|
|
class FakeHaReader(HaReader):
|
|
def __init__(self, entities: list[HaEntitySummary], history: dict[str, list[float]]) -> None:
|
|
self._entities = entities
|
|
self._history = history
|
|
self.read_entities_calls = 0
|
|
self.service_calls: list[tuple[str, str, dict[str, object]]] = []
|
|
|
|
def read_entities(self) -> list[HaEntitySummary]:
|
|
self.read_entities_calls += 1
|
|
return list(self._entities)
|
|
|
|
def discover(
|
|
self,
|
|
domains: set[str] | None = None,
|
|
learnable: bool | None = None,
|
|
) -> list[DiscoveredEntity]:
|
|
return discover_entities(self._entities, domains=domains, learnable=learnable)
|
|
|
|
def read_history(
|
|
self,
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> list[EntityHistorySeries]:
|
|
base = start_time
|
|
return [
|
|
EntityHistorySeries(
|
|
entity_id=entity_id,
|
|
points=[
|
|
NumericHistoryPoint(
|
|
timestamp=base + timedelta(hours=index),
|
|
value=value,
|
|
)
|
|
for index, value in enumerate(self._history.get(entity_id, []))
|
|
],
|
|
)
|
|
for entity_id in entity_ids
|
|
if entity_id in self._history
|
|
]
|
|
|
|
def read_state_history(
|
|
self,
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> list[StateHistorySeries]:
|
|
return []
|
|
|
|
def read_logbook(
|
|
self,
|
|
entity_id: str,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> list[LogbookEntry]:
|
|
return []
|
|
|
|
def call_service(
|
|
self,
|
|
domain: str,
|
|
service: str,
|
|
service_data: dict[str, object],
|
|
) -> list[object]:
|
|
self.service_calls.append((domain, service, service_data))
|
|
return []
|
|
|
|
def find_automations_for_entity(
|
|
self,
|
|
entity_id: str,
|
|
) -> list[HaAutomationSummary]:
|
|
return []
|
|
|
|
|
|
def _install_service(tmp_path: Path) -> None:
|
|
entities = [
|
|
HaEntitySummary(
|
|
entity_id="light.abstellkammer",
|
|
domain="light",
|
|
friendly_name="Abstellkammer Licht",
|
|
area_name="Abstellkammer",
|
|
),
|
|
HaEntitySummary(
|
|
entity_id="sensor.abstellkammer_illuminance",
|
|
domain="sensor",
|
|
device_class="illuminance",
|
|
state_class="measurement",
|
|
unit_of_measurement="lx",
|
|
friendly_name="Abstellkammer Helligkeit",
|
|
area_name="Abstellkammer",
|
|
state="12",
|
|
),
|
|
HaEntitySummary(
|
|
entity_id="binary_sensor.abstellkammer_motion",
|
|
domain="binary_sensor",
|
|
device_class="motion",
|
|
friendly_name="Abstellkammer Bewegung",
|
|
area_name="Abstellkammer",
|
|
state="off",
|
|
),
|
|
HaEntitySummary(
|
|
entity_id="sensor.pfsense_interface_vpn_inbytes",
|
|
domain="sensor",
|
|
device_class="data_size",
|
|
state_class="measurement",
|
|
unit_of_measurement="KiB",
|
|
friendly_name="pfSense Interface VPN inbytes",
|
|
),
|
|
]
|
|
settings = Settings(
|
|
ha_url="http://ha.local",
|
|
ha_token="token",
|
|
model_store=str(tmp_path / "models"),
|
|
automation_store=str(tmp_path / "automations"),
|
|
actuator_store=str(tmp_path / "actuators"),
|
|
history_days=14,
|
|
min_training_points=5,
|
|
retrain_stale_hours=24,
|
|
reconcile_interval_seconds=900,
|
|
)
|
|
app.state.registry = ModelRegistry(tmp_path / "models")
|
|
app.state.actuator_store = ActuatorStore(tmp_path / "actuators")
|
|
app.state.dashboard_cache = DashboardCache(tmp_path / "actuators" / "dashboard_cache.sqlite3")
|
|
app.state.ha_reader = FakeHaReader(
|
|
entities,
|
|
{"sensor.abstellkammer_illuminance": [10, 11, 12, 13, 14, 15]},
|
|
)
|
|
app.state.actuator_service = ActuatorReconciliationService(
|
|
ha_reader=app.state.ha_reader,
|
|
store=app.state.actuator_store,
|
|
registry=app.state.registry,
|
|
settings=settings,
|
|
)
|
|
app.state.behavior_engine = BehaviorEngine(
|
|
ha_reader=app.state.ha_reader,
|
|
store=app.state.actuator_store,
|
|
settings=settings,
|
|
)
|
|
|
|
|
|
def test_actuator_api_configures_reconciles_and_removes(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
|
|
created = client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
assert created.status_code == 201
|
|
assert created.json()["assignment"]["selected_numeric_entity_id"] == (
|
|
"sensor.abstellkammer_illuminance"
|
|
)
|
|
|
|
listed = client.get("/v1/actuators")
|
|
assert listed.status_code == 200
|
|
assert listed.json()[0]["lifecycle"]["status"] == "trained"
|
|
assert listed.json()[0]["behavior"]["mode"] == "shadow"
|
|
|
|
evaluation = client.post("/v1/actuators/light.abstellkammer/evaluate")
|
|
assert evaluation.status_code == 200
|
|
|
|
premature_activation = client.post(
|
|
"/v1/actuators/light.abstellkammer/activation",
|
|
json={"active": True},
|
|
)
|
|
assert premature_activation.status_code == 409
|
|
|
|
reconciliation = client.post("/v1/actuators/reconciliation/run")
|
|
assert reconciliation.status_code == 200
|
|
assert reconciliation.json()["trained_models"] == 1
|
|
|
|
removed = client.delete("/v1/actuators/light.abstellkammer")
|
|
assert removed.status_code == 204
|
|
assert client.get("/v1/actuators").json() == []
|
|
|
|
|
|
def test_manual_assignment_endpoint_updates_context(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
response = client.post(
|
|
"/v1/actuators/light.abstellkammer/assignment",
|
|
json={
|
|
"numeric_entity_id": "sensor.abstellkammer_illuminance",
|
|
"context_entity_ids": ["binary_sensor.abstellkammer_motion"],
|
|
"note": "Manuell gesetzt",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["assignment"]["source"] == "manual"
|
|
assert payload["assignment"]["selected_numeric_entity_id"] == (
|
|
"sensor.abstellkammer_illuminance"
|
|
)
|
|
assert payload["assignment"]["selected_context_entity_ids"] == [
|
|
"binary_sensor.abstellkammer_motion"
|
|
]
|
|
|
|
|
|
def test_weight_override_endpoint_updates_sensor_relevance(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
client.post(
|
|
"/v1/actuators/light.abstellkammer/assignment",
|
|
json={
|
|
"numeric_entity_id": "sensor.abstellkammer_illuminance",
|
|
"context_entity_ids": ["binary_sensor.abstellkammer_motion"],
|
|
},
|
|
)
|
|
|
|
response = client.post(
|
|
"/v1/actuators/light.abstellkammer/weights",
|
|
json={
|
|
"sensor_weights": {
|
|
"sensor.abstellkammer_illuminance": 0.75,
|
|
"binary_sensor.abstellkammer_motion": 0.5,
|
|
},
|
|
"sensor_weight_groups": [
|
|
{
|
|
"group_id": "abstellkammer_context",
|
|
"name": "Abstellkammer Kontext",
|
|
"entity_ids": [
|
|
"sensor.abstellkammer_illuminance",
|
|
"binary_sensor.abstellkammer_motion",
|
|
],
|
|
"weight": 0.8,
|
|
}
|
|
],
|
|
"note": "Gewichtung korrigiert",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["manual_override"]["sensor_weights"]["sensor.abstellkammer_illuminance"] == 0.75
|
|
assert payload["manual_override"]["sensor_weight_groups"][0]["group_id"] == (
|
|
"abstellkammer_context"
|
|
)
|
|
numeric = {
|
|
candidate["entity_id"]: candidate
|
|
for candidate in payload["numeric_candidates"]
|
|
}
|
|
assert numeric["sensor.abstellkammer_illuminance"]["manual_weight"] == 0.75
|
|
assert numeric["sensor.abstellkammer_illuminance"]["effective_weight"] == 0.75
|
|
|
|
|
|
def test_actuator_simulation_ranks_sensor_states_without_switching(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
client.post(
|
|
"/v1/actuators/light.abstellkammer/assignment",
|
|
json={
|
|
"numeric_entity_id": "sensor.abstellkammer_illuminance",
|
|
"context_entity_ids": ["binary_sensor.abstellkammer_motion"],
|
|
},
|
|
)
|
|
store = app.state.actuator_store
|
|
record = store.get("light.abstellkammer")
|
|
now = datetime.now(timezone.utc)
|
|
local = now.astimezone(ZoneInfo("Europe/Berlin"))
|
|
local_minute = local.hour * 60 + local.minute
|
|
patterns = [
|
|
BehaviorPattern(
|
|
target_state="on",
|
|
minute_of_day=local_minute,
|
|
weekday=now.weekday(),
|
|
context_states={
|
|
"sensor.abstellkammer_illuminance": "12",
|
|
"binary_sensor.abstellkammer_motion": "on",
|
|
},
|
|
source="user",
|
|
weight=1.0,
|
|
observed_at=now,
|
|
)
|
|
for _ in range(3)
|
|
]
|
|
patterns.extend(
|
|
[
|
|
BehaviorPattern(
|
|
target_state="off",
|
|
minute_of_day=local_minute,
|
|
weekday=now.weekday(),
|
|
context_states={
|
|
"sensor.abstellkammer_illuminance": "12",
|
|
"binary_sensor.abstellkammer_motion": "off",
|
|
},
|
|
source="user",
|
|
weight=0.5,
|
|
observed_at=now,
|
|
)
|
|
for _ in range(3)
|
|
]
|
|
)
|
|
store.upsert(
|
|
record.model_copy(
|
|
update={
|
|
"behavior": record.behavior.model_copy(
|
|
update={
|
|
"patterns": patterns,
|
|
"sample_count": len(patterns),
|
|
"high_confidence_sample_count": len(patterns),
|
|
"activation_ready": True,
|
|
"activation_reason": "Testfreigabe.",
|
|
}
|
|
)
|
|
}
|
|
)
|
|
)
|
|
|
|
response = client.post(
|
|
"/v1/actuators/light.abstellkammer/simulate",
|
|
json={
|
|
"state_options": {"binary_sensor.abstellkammer_motion": ["off", "on"]},
|
|
"sensor_weights": {
|
|
"binary_sensor.abstellkammer_motion": 1.0,
|
|
"sensor.abstellkammer_illuminance": 0.25,
|
|
},
|
|
"max_results": 2,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert len(payload) == 2
|
|
assert payload[0]["prediction"]["target_state"] == "on"
|
|
assert payload[0]["sensor_states"]["binary_sensor.abstellkammer_motion"] == "on"
|
|
assert payload[0]["sensor_weights"]["sensor.abstellkammer_illuminance"] == 0.25
|
|
assert app.state.ha_reader.service_calls == []
|
|
|
|
|
|
def test_safety_profile_can_block_actuator_manually(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
response = client.post(
|
|
"/v1/actuators/light.abstellkammer/safety",
|
|
json={
|
|
"safety": {
|
|
"stage": "shadow",
|
|
"manual_block": True,
|
|
"min_confidence": 0.9,
|
|
"cooldown_seconds": 120,
|
|
"rules": [
|
|
{
|
|
"rule_id": "manual_block",
|
|
"label": "Manuelle Sperre respektieren",
|
|
"enabled": True,
|
|
"blocking": True,
|
|
"reason": "Test",
|
|
}
|
|
],
|
|
"note": "Test",
|
|
}
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["behavior"]["safety"]["manual_block"] is True
|
|
assert payload["behavior"]["safety"]["min_confidence"] == 0.9
|
|
assert payload["behavior"]["safety"]["cooldown_seconds"] == 120
|
|
|
|
|
|
def test_feedback_adapts_sensor_weights_and_model_can_rollback(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post(
|
|
"/v1/actuators",
|
|
json={"actuator_entity_id": "light.abstellkammer"},
|
|
)
|
|
record = app.state.actuator_store.get("light.abstellkammer")
|
|
version_id = "model-test"
|
|
snapshot = ModelSnapshot(
|
|
version_id=version_id,
|
|
sample_count=1,
|
|
high_confidence_sample_count=1,
|
|
average_confidence=0.9,
|
|
patterns=[],
|
|
reason="Test-Snapshot",
|
|
)
|
|
app.state.actuator_store.upsert(
|
|
record.model_copy(
|
|
update={
|
|
"behavior": record.behavior.model_copy(
|
|
update={
|
|
"model_snapshots": [snapshot],
|
|
"active_model_version": "model-current",
|
|
"sample_count": 2,
|
|
}
|
|
)
|
|
}
|
|
)
|
|
)
|
|
|
|
feedback = client.post(
|
|
"/v1/actuators/light.abstellkammer/feedback",
|
|
json={"correct": False, "expected_state": "off"},
|
|
)
|
|
rollback = client.post(
|
|
"/v1/actuators/light.abstellkammer/model/rollback",
|
|
json={"version_id": version_id},
|
|
)
|
|
|
|
assert feedback.status_code == 200
|
|
feedback_payload = feedback.json()
|
|
assert feedback_payload["behavior"]["adaptive_weight_updates"]
|
|
assert feedback_payload["manual_override"]["sensor_weights"]
|
|
assert rollback.status_code == 200
|
|
assert rollback.json()["behavior"]["active_model_version"] == version_id
|
|
|
|
|
|
def test_feedback_never_automate_sets_manual_block(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post(
|
|
"/v1/actuators",
|
|
json={"actuator_entity_id": "light.abstellkammer"},
|
|
)
|
|
|
|
feedback = client.post(
|
|
"/v1/actuators/light.abstellkammer/feedback",
|
|
json={"correct": False, "kind": "never_automate"},
|
|
)
|
|
|
|
assert feedback.status_code == 200
|
|
payload = feedback.json()
|
|
assert payload["behavior"]["safety"]["manual_block"] is True
|
|
assert payload["behavior"]["feedback_log"][-1] == "never_automate"
|
|
|
|
|
|
def test_backup_export_restore_and_planning_refresh(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
backup = client.get("/v1/actuators/backup/export")
|
|
dry_run = client.post(
|
|
"/v1/actuators/light.abstellkammer/dry-run",
|
|
json={"enabled": True},
|
|
)
|
|
planning = client.post("/v1/actuators/planning/refresh")
|
|
restore = client.post(
|
|
"/v1/actuators/backup/restore",
|
|
json={"backup": backup.json(), "replace_existing": True},
|
|
)
|
|
|
|
assert backup.status_code == 200
|
|
assert backup.json()["records"][0]["actuator_entity_id"] == "light.abstellkammer"
|
|
assert dry_run.status_code == 200
|
|
assert dry_run.json()["behavior"]["dry_run_enabled"] is True
|
|
assert planning.status_code == 200
|
|
assert "agent_insights" in planning.json()[0]["behavior"]
|
|
assert restore.status_code == 200
|
|
assert restore.json()["restored_records"] == 1
|
|
|
|
|
|
def test_summary_is_lightweight_and_uses_cached_entity_metadata(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.get("/v1/actuators/discovery")
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
response = client.get("/v1/actuators/summary")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload[0]["actuator_entity_id"] == "light.abstellkammer"
|
|
assert payload[0]["friendly_name"] == "Abstellkammer Licht"
|
|
assert payload[0]["area_name"] == "Abstellkammer"
|
|
assert "behavior" not in payload[0]
|
|
assert "numeric_candidates" not in payload[0]
|
|
|
|
|
|
def test_dashboard_overview_uses_cache_without_ha_roundtrip(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
reader = app.state.ha_reader
|
|
client.get("/v1/actuators/discovery")
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
calls_before = reader.read_entities_calls
|
|
|
|
response = client.get("/v1/actuators/dashboard")
|
|
|
|
assert response.status_code == 200
|
|
assert reader.read_entities_calls == calls_before
|
|
payload = response.json()
|
|
assert payload["cache"]["available"] is True
|
|
assert payload["cache"]["entity_count"] == 4
|
|
assert payload["actuators"][0]["friendly_name"] == "Abstellkammer Licht"
|
|
assert payload["discovery_groups"]
|
|
assert payload["jobs"]["jobs"][-1]["kind"] == "discovery"
|
|
|
|
|
|
def test_reconciliation_run_records_visible_job_queue(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
response = client.post("/v1/actuators/reconciliation/run")
|
|
jobs = client.get("/v1/actuators/job-queue/state")
|
|
|
|
assert response.status_code == 200
|
|
assert jobs.status_code == 200
|
|
payload = jobs.json()
|
|
assert [job["kind"] for job in payload["jobs"][-3:]] == [
|
|
"reconciliation",
|
|
"training",
|
|
"evaluation",
|
|
]
|
|
assert payload["jobs"][-1]["status"] == "completed"
|
|
|
|
|
|
def test_dashboard_start_path_stays_within_three_second_budget(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.get("/v1/actuators/discovery")
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
root_started_at = perf_counter()
|
|
root_response = client.get("/")
|
|
root_elapsed = perf_counter() - root_started_at
|
|
|
|
dashboard_started_at = perf_counter()
|
|
dashboard_response = client.get("/v1/actuators/dashboard/start")
|
|
dashboard_elapsed = perf_counter() - dashboard_started_at
|
|
|
|
assert root_response.status_code == 200
|
|
assert dashboard_response.status_code == 200
|
|
assert root_elapsed < 3.0
|
|
assert dashboard_elapsed < 3.0
|
|
|
|
|
|
def test_dashboard_reports_performance_budget_and_anomalies(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.get("/v1/actuators/discovery")
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
store = app.state.actuator_store
|
|
job = store.start_job(kind="training", trigger="test", summary="Langsamer Testjob")
|
|
queue = store.load_job_queue()
|
|
queue.jobs = [
|
|
item.model_copy(update={"started_at": datetime.now(timezone.utc) - timedelta(seconds=4)})
|
|
if item.job_id == job.job_id
|
|
else item
|
|
for item in queue.jobs
|
|
]
|
|
store._persist_job_queue(queue)
|
|
store.finish_job(job.job_id, status=JobStatus.COMPLETED, summary="Fertig")
|
|
|
|
dashboard_response = client.get("/v1/actuators/dashboard")
|
|
start_response = client.get("/v1/actuators/dashboard/start")
|
|
system_response = client.get("/v1/actuators/dashboard/system")
|
|
anomalies_response = client.get("/v1/actuators/anomalies")
|
|
|
|
assert dashboard_response.status_code == 200
|
|
assert start_response.status_code == 200
|
|
assert system_response.status_code == 200
|
|
system = dashboard_response.json()["system"]
|
|
start_payload = start_response.json()
|
|
assert start_payload["jobs"]["jobs"] == []
|
|
assert start_payload["discovery_groups"] == []
|
|
assert system_response.json()["actuators"] == []
|
|
assert system["performance_budget_ms"] == 3000
|
|
assert system["slow_job_count"] == 1
|
|
assert system["performance_status"] == "slow"
|
|
assert system["anomaly_count"] >= 1
|
|
assert anomalies_response.status_code == 200
|
|
assert anomalies_response.json()
|
|
|
|
|
|
def test_dashboard_system_and_start_do_not_materialize_entity_cache(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.get("/v1/actuators/discovery")
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
def fail_full_payload(self: DashboardCache) -> dict[str, object]:
|
|
raise AssertionError("full entity payload must not be loaded")
|
|
|
|
monkeypatch.setattr(DashboardCache, "load_entities_payload", fail_full_payload)
|
|
|
|
system_response = client.get("/v1/actuators/dashboard/system")
|
|
start_response = client.get("/v1/actuators/dashboard/start")
|
|
|
|
assert system_response.status_code == 200
|
|
assert system_response.json()["actuators"] == []
|
|
assert system_response.json()["cache"]["entity_count"] == 4
|
|
assert start_response.status_code == 200
|
|
assert start_response.json()["actuators"][0]["friendly_name"] == "Abstellkammer Licht"
|
|
|
|
|
|
def test_actuator_detail_uses_compact_payload(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
client.get("/v1/actuators/discovery")
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
|
|
response = client.get("/v1/actuators/light.abstellkammer/detail")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["behavior"]["patterns"] == []
|
|
assert all(
|
|
snapshot["patterns"] == []
|
|
for snapshot in payload["behavior"]["model_snapshots"]
|
|
)
|
|
|
|
|
|
def test_discovery_reads_entities_once_and_reuses_them(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
reader = app.state.ha_reader
|
|
|
|
response = client.get("/v1/actuators/discovery", params={"refresh": True})
|
|
|
|
assert response.status_code == 200
|
|
assert reader.read_entities_calls == 1
|
|
|
|
|
|
def test_context_options_returns_learnable_entities(tmp_path: Path) -> None:
|
|
with TestClient(app) as client:
|
|
_install_service(tmp_path)
|
|
|
|
client.post("/v1/actuators", json={"actuator_entity_id": "light.abstellkammer"})
|
|
response = client.get(
|
|
"/v1/actuators/context-options",
|
|
params={"actuator_entity_id": "light.abstellkammer"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
entity_ids = {item["entity_id"] for item in response.json()}
|
|
assert "sensor.abstellkammer_illuminance" in entity_ids
|
|
assert "binary_sensor.abstellkammer_motion" in entity_ids
|
|
assert "sensor.pfsense_interface_vpn_inbytes" not in entity_ids
|
|
|
|
|
|
def test_actuator_discovery_prefers_light_over_duplicate_switch() -> None:
|
|
entities = {
|
|
"light.schreibtisch": HaEntitySummary(
|
|
entity_id="light.schreibtisch",
|
|
domain="light",
|
|
friendly_name="Schreibtisch Licht",
|
|
device_id="device-1",
|
|
),
|
|
"switch.schreibtisch": HaEntitySummary(
|
|
entity_id="switch.schreibtisch",
|
|
domain="switch",
|
|
friendly_name="Schreibtisch Schalter",
|
|
device_id="device-1",
|
|
),
|
|
"cover.rollladen": HaEntitySummary(
|
|
entity_id="cover.rollladen",
|
|
domain="cover",
|
|
friendly_name="Rollladen",
|
|
device_id="device-2",
|
|
),
|
|
}
|
|
|
|
result = _deduplicate_actuator_ids(
|
|
[
|
|
("switch.schreibtisch", "switch_socket"),
|
|
("light.schreibtisch", "light"),
|
|
("cover.rollladen", "cover_shutter"),
|
|
],
|
|
entities,
|
|
)
|
|
|
|
assert result == ["cover.rollladen", "light.schreibtisch"]
|