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_room_management_overview_groups_actuators_with_sensors_and_rules(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/settings/rooms") assert response.status_code == 200 payload = response.json() room = payload["rooms"][0] assert room["room"] == "Abstellkammer" assert room["actuator_count"] == 1 actuator = room["actuators"][0] assert actuator["actuator_entity_id"] == "light.abstellkammer" assert actuator["sensors"] assert actuator["prediction_rules"] assert any(sensor["entity_id"] == "binary_sensor.abstellkammer_motion" for sensor in room["sensors"]) 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"]