from __future__ import annotations from time import perf_counter from datetime import datetime, timedelta, timezone from pathlib import Path from fastapi.testclient import TestClient from app.actuators.cache_db import DashboardCache from app.actuators.lifecycle import ActuatorReconciliationService from app.actuators.models import JobStatus, ModelSnapshot from app.actuators.store import ActuatorStore from app.behavior.engine import BehaviorEngine from app.config import Settings from app.api.v1.actuators import _deduplicate_actuator_ids 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_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_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_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"]