from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from app.actuators.models import ( BehaviorMode, BehaviorPattern, BehaviorState, BehaviorStatus, ExecutionEvent, ) from app.actuators.store import ActuatorStore from app.behavior.engine import BehaviorEngine, predict_behavior, service_for_state from app.config import Settings from app.ha.history import ( LogbookEntry, StateHistoryPoint, StateHistorySeries, ) from app.ha.models import HaAutomationSummary, HaEntitySummary from app.ha.reader import HaReader class FakeBehaviorReader(HaReader): def __init__( self, *, entities: list[HaEntitySummary], history: list[StateHistorySeries], logbook: list[LogbookEntry], ) -> None: self.entities = entities self.history = history self.logbook = logbook self.service_calls: list[tuple[str, str, dict[str, object]]] = [] self.automations: list[HaAutomationSummary] = [] def read_entities(self) -> list[HaEntitySummary]: return list(self.entities) def read_state_history( self, entity_ids: list[str], start_time: datetime, end_time: datetime, ) -> list[StateHistorySeries]: return [series for series in self.history if series.entity_id in entity_ids] def read_logbook( self, entity_id: str, start_time: datetime, end_time: datetime, ) -> list[LogbookEntry]: return [entry for entry in self.logbook if entry.entity_id == entity_id] 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 list(self.automations) def _settings(tmp_path: Path) -> Settings: return Settings( actuator_store=str(tmp_path / "actuators"), model_store=str(tmp_path / "models"), automation_store=str(tmp_path / "automations"), history_days=14, min_behavior_actions=3, prediction_confidence=0.8, prediction_window_minutes=30, execution_cooldown_seconds=900, timezone="Europe/Berlin", ) def _reader(now: datetime) -> FakeBehaviorReader: actuator_points: list[StateHistoryPoint] = [] logbook: list[LogbookEntry] = [] for days_ago in (3, 2, 1): action_at = now - timedelta(days=days_ago) actuator_points.extend( [ StateHistoryPoint(timestamp=action_at - timedelta(minutes=1), state="off"), StateHistoryPoint(timestamp=action_at, state="on"), StateHistoryPoint(timestamp=action_at + timedelta(hours=6), state="off"), ] ) logbook.extend( [ LogbookEntry( entity_id="light.office", timestamp=action_at, message="turned on", context_user_id="user-1", ), LogbookEntry( entity_id="light.office", timestamp=action_at + timedelta(hours=6), message="turned off", context_domain="automation", context_service="trigger", ), ] ) actuator_points.sort(key=lambda point: point.timestamp) context_points = [ StateHistoryPoint(timestamp=now - timedelta(days=7), state="on"), ] return FakeBehaviorReader( entities=[ HaEntitySummary(entity_id="light.office", domain="light", state="off"), HaEntitySummary( entity_id="binary_sensor.office_presence", domain="binary_sensor", state="on", ), ], history=[ StateHistorySeries(entity_id="light.office", points=actuator_points), StateHistorySeries( entity_id="binary_sensor.office_presence", points=context_points, ), ], logbook=logbook, ) def _engine(tmp_path: Path, now: datetime) -> tuple[BehaviorEngine, FakeBehaviorReader]: settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) record = store.configure("light.office") store.upsert( record.model_copy( update={ "assignment": record.assignment.model_copy( update={ "selected_context_entity_ids": [ "binary_sensor.office_presence" ], } ) } ) ) reader = _reader(now) return ( BehaviorEngine(ha_reader=reader, store=store, settings=settings), reader, ) def test_engine_trains_predicts_in_shadow_and_executes_only_after_approval( tmp_path: Path, ) -> None: now = datetime.now(timezone.utc).replace(second=0, microsecond=0) engine, reader = _engine(tmp_path, now) trained = engine.train("light.office") shadow = engine.evaluate("light.office") assert trained.behavior.status is BehaviorStatus.TRAINED assert trained.behavior.sample_count == 6 assert trained.behavior.high_confidence_sample_count == 6 assert shadow.behavior.mode is BehaviorMode.SHADOW assert shadow.behavior.prediction is not None assert shadow.behavior.prediction.target_state == "on" assert reader.service_calls == [] engine.set_active("light.office", active=True) active = engine.evaluate("light.office") assert active.behavior.mode is BehaviorMode.ACTIVE assert active.behavior.prediction is not None assert active.behavior.prediction.executed is True assert reader.service_calls == [ ("light", "turn_on", {"entity_id": "light.office"}) ] def test_engine_counts_known_automation_actions_like_manual_actions( tmp_path: Path, ) -> None: now = datetime.now(timezone.utc).replace(second=0, microsecond=0) engine, _ = _engine(tmp_path, now) trained = engine.train("light.office") assert {pattern.target_state for pattern in trained.behavior.patterns} == { "on", "off", } assert {pattern.source for pattern in trained.behavior.patterns} == { "user", "automation", } assert trained.behavior.high_confidence_sample_count == 6 assert {pattern.weight for pattern in trained.behavior.patterns} == {1.0} def test_engine_learns_causal_automation_with_activation_credit( tmp_path: Path, ) -> None: now = datetime.now(timezone.utc).replace(second=0, microsecond=0) actuator_points: list[StateHistoryPoint] = [] door_points: list[StateHistoryPoint] = [] logbook: list[LogbookEntry] = [] for days_ago in (3, 2, 1): action_at = now - timedelta(days=days_ago) actuator_points.extend( [ StateHistoryPoint( timestamp=action_at - timedelta(minutes=1), state="off", ), StateHistoryPoint(timestamp=action_at, state="on"), ] ) door_points.extend( [ StateHistoryPoint( timestamp=action_at - timedelta(minutes=1), state="off", ), StateHistoryPoint( timestamp=action_at - timedelta(seconds=1), state="on", ), ] ) logbook.append( LogbookEntry( entity_id="light.storage", timestamp=action_at, message="turned on", context_domain="automation", context_service="trigger", ) ) actuator_points.sort(key=lambda point: point.timestamp) door_points.sort(key=lambda point: point.timestamp) settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) record = store.configure("light.storage") store.upsert( record.model_copy( update={ "assignment": record.assignment.model_copy( update={ "selected_context_entity_ids": [ "binary_sensor.storage_door" ], } ) } ) ) reader = FakeBehaviorReader( entities=[], history=[ StateHistorySeries( entity_id="light.storage", points=actuator_points, ), StateHistorySeries( entity_id="binary_sensor.storage_door", points=door_points, ), ], logbook=logbook, ) engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings) trained = engine.train("light.storage") automation_patterns = [ pattern for pattern in trained.behavior.patterns if pattern.source == "automation" ] assert len(automation_patterns) == 3 assert trained.behavior.high_confidence_sample_count == 3 assert {pattern.weight for pattern in automation_patterns} == {1.0} assert { ( pattern.trigger_entity_id, pattern.trigger_from_state, pattern.trigger_to_state, ) for pattern in automation_patterns } == {("binary_sensor.storage_door", "off", "on")} active = engine.set_active("light.storage", active=True) assert active.behavior.mode is BehaviorMode.ACTIVE def test_active_mode_rejects_unsafe_domains(tmp_path: Path) -> None: settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) record = store.configure("lock.front_door") store.upsert( record.model_copy( update={ "behavior": record.behavior.model_copy( update={"status": BehaviorStatus.TRAINED} ) } ) ) reader = FakeBehaviorReader(entities=[], history=[], logbook=[]) engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings) with pytest.raises(ValueError, match="nicht freigegeben"): engine.set_active("lock.front_door", active=True) def test_active_mode_requires_trusted_manual_or_automation_actions(tmp_path: Path) -> None: settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) record = store.configure("light.office") store.upsert( record.model_copy( update={ "behavior": record.behavior.model_copy( update={ "status": BehaviorStatus.TRAINED, "sample_count": 3, "high_confidence_sample_count": 0, } ) } ) ) reader = FakeBehaviorReader(entities=[], history=[], logbook=[]) engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings) with pytest.raises(ValueError, match="Freigabe"): engine.set_active("light.office", active=True) def test_control_handoff_pauses_and_restores_matching_automation( tmp_path: Path, ) -> None: settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) record = store.configure("light.storage") store.upsert( record.model_copy( update={ "behavior": record.behavior.model_copy( update={ "status": BehaviorStatus.TRAINED, "sample_count": 3, "high_confidence_sample_count": 3, "activation_ready": True, "activation_reason": "Freigabe bereit.", } ) } ) ) reader = FakeBehaviorReader(entities=[], history=[], logbook=[]) reader.automations = [ HaAutomationSummary( entity_id="automation.storage_light", config_id="123", friendly_name="Storage light", enabled=True, ) ] engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings) active = engine.set_active( "light.storage", active=True, pause_matching_automations=True, ) shadow = engine.set_active( "light.storage", active=False, restore_paused_automations=True, ) assert active.behavior.mode is BehaviorMode.ACTIVE assert active.behavior.paused_automation_entity_ids == [ "automation.storage_light" ] assert shadow.behavior.mode is BehaviorMode.SHADOW assert shadow.behavior.paused_automation_entity_ids == [] assert reader.service_calls == [ ( "automation", "turn_off", {"entity_id": "automation.storage_light"}, ), ( "automation", "turn_on", {"entity_id": "automation.storage_light"}, ), ] def test_cooldown_allows_opposite_follow_up_action(tmp_path: Path) -> None: settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) reader = FakeBehaviorReader(entities=[], history=[], logbook=[]) engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings) now = datetime.now(timezone.utc) behavior = BehaviorState( mode=BehaviorMode.ACTIVE, last_executed_at=now - timedelta(seconds=5), execution_events=[ ExecutionEvent(target_state="on", executed_at=now - timedelta(seconds=5)) ], ) assert engine._cooldown_elapsed(behavior, now, "off") is True assert engine._cooldown_elapsed(behavior, now, "on") is False @pytest.mark.parametrize( ("domain", "state", "service"), [ ("light", "on", "turn_on"), ("switch", "off", "turn_off"), ("cover", "open", "open_cover"), ("cover", "closed", "close_cover"), ("lock", "unlocked", None), ], ) def test_service_for_state_is_strictly_allowlisted( domain: str, state: str, service: str | None, ) -> None: assert service_for_state(domain, state) == service def test_prediction_requires_temporal_support() -> None: assert predict_behavior( [], current_context={}, now=datetime.now(timezone.utc), min_support=3, window_minutes=30, ) is None def test_prediction_uses_fresh_causal_context_transition_outside_time_window() -> None: now = datetime.now(timezone.utc).replace(second=0, microsecond=0) patterns = [ BehaviorPattern( target_state="on", minute_of_day=60, weekday=0, context_states={"binary_sensor.storage_door": "on"}, trigger_entity_id="binary_sensor.storage_door", trigger_from_state="off", trigger_to_state="on", source="automation", weight=0.7, observed_at=now - timedelta(days=days_ago), ) for days_ago in (3, 2, 1) ] prediction = predict_behavior( patterns, current_context={"binary_sensor.storage_door": "on"}, current_context_changed_at={ "binary_sensor.storage_door": now - timedelta(seconds=10) }, now=now, min_support=3, window_minutes=30, ) assert prediction is not None assert prediction.target_state == "on" assert prediction.matching_patterns == 3 assert prediction.confidence == 0.7 assert "frischen Sensorwechsel" in prediction.reason def test_prediction_ignores_stale_causal_context_state() -> None: now = datetime.now(timezone.utc).replace(second=0, microsecond=0) pattern = BehaviorPattern( target_state="on", minute_of_day=60, weekday=0, context_states={"binary_sensor.storage_door": "on"}, trigger_entity_id="binary_sensor.storage_door", trigger_from_state="off", trigger_to_state="on", source="automation", weight=0.7, observed_at=now - timedelta(days=1), ) assert predict_behavior( [pattern], current_context={"binary_sensor.storage_door": "on"}, current_context_changed_at={ "binary_sensor.storage_door": now - timedelta(minutes=5) }, now=now, min_support=1, window_minutes=30, ) is None def test_state_change_uses_websocket_context_state_for_immediate_action( tmp_path: Path, ) -> None: now = datetime.now(timezone.utc).replace(microsecond=0) settings = _settings(tmp_path) store = ActuatorStore(settings.actuator_store) record = store.configure("light.storage") record = record.model_copy( update={ "assignment": record.assignment.model_copy( update={ "selected_context_entity_ids": ["binary_sensor.storage_door"], } ), "behavior": record.behavior.model_copy( update={ "mode": BehaviorMode.ACTIVE, "status": BehaviorStatus.TRAINED, "activation_ready": True, "patterns": [ BehaviorPattern( target_state="on", minute_of_day=60, weekday=0, context_states={"binary_sensor.storage_door": "on"}, trigger_entity_id="binary_sensor.storage_door", trigger_from_state="off", trigger_to_state="on", source="automation", weight=1.0, observed_at=now - timedelta(days=days_ago), ) for days_ago in (3, 2, 1) ], } ), } ) store.upsert(record) reader = FakeBehaviorReader( entities=[ HaEntitySummary(entity_id="light.storage", domain="light", state="off"), HaEntitySummary( entity_id="binary_sensor.storage_door", domain="binary_sensor", state="off", last_changed=now - timedelta(minutes=5), ), ], history=[], logbook=[], ) engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings) engine.handle_state_change( "binary_sensor.storage_door", {"state": "on", "last_changed": now.isoformat()}, ) assert reader.service_calls == [ ("light", "turn_on", {"entity_id": "light.storage"}) ]