262 lines
8.3 KiB
Python
262 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.actuators.models import BehaviorMode, BehaviorStatus
|
|
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 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]]] = []
|
|
|
|
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 _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 == 3
|
|
assert trained.behavior.high_confidence_sample_count == 3
|
|
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_excludes_known_automation_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"}
|
|
assert {pattern.source for pattern in trained.behavior.patterns} == {"user"}
|
|
|
|
|
|
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_user_attributed_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="eindeutig dir zugeordnete"):
|
|
engine.set_active("light.office", active=True)
|
|
|
|
|
|
@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
|