156 lines
4.7 KiB
Python
156 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.event_core import EventCore
|
|
from app.core.handoff import HandoffMatrix
|
|
from app.core.models import (
|
|
BehaviorPatternV2,
|
|
ControlProfile,
|
|
Decision,
|
|
EntityState,
|
|
HandoffMode,
|
|
LearningProfile,
|
|
LearningState,
|
|
RuntimeState,
|
|
SafetyStage,
|
|
StateEvent,
|
|
)
|
|
from app.core.stores import FutureStores
|
|
|
|
|
|
def test_event_core_routes_state_event_to_decision(tmp_path: Path) -> None:
|
|
stores = FutureStores(tmp_path)
|
|
stores.save_runtime(
|
|
RuntimeState(
|
|
entities={
|
|
"light.storage": EntityState(
|
|
entity_id="light.storage",
|
|
domain="light",
|
|
state="off",
|
|
)
|
|
}
|
|
)
|
|
)
|
|
stores.save_learning(
|
|
LearningState(
|
|
profiles={
|
|
"light.storage": LearningProfile(
|
|
actuator_entity_id="light.storage",
|
|
patterns=[
|
|
BehaviorPatternV2(
|
|
actuator_entity_id="light.storage",
|
|
target_state="on",
|
|
trigger_entity_id="binary_sensor.storage_door",
|
|
trigger_state="on",
|
|
support=3,
|
|
confidence=0.95,
|
|
)
|
|
],
|
|
)
|
|
}
|
|
)
|
|
)
|
|
stores.save_control(
|
|
stores.control().model_copy(
|
|
update={
|
|
"profiles": {
|
|
"light.storage": ControlProfile(
|
|
actuator_entity_id="light.storage",
|
|
stage=SafetyStage.DRY_RUN,
|
|
min_confidence=0.8,
|
|
)
|
|
}
|
|
}
|
|
)
|
|
)
|
|
core = EventCore(stores)
|
|
|
|
audit = core.process_state_event(
|
|
StateEvent(entity_id="binary_sensor.storage_door", new_state="on")
|
|
)
|
|
|
|
decisions = [item for item in audit if item.kind == "decision"]
|
|
assert decisions
|
|
assert decisions[0].decision is not None
|
|
assert decisions[0].decision.target_state == "on"
|
|
assert decisions[0].decision.allowed is True
|
|
assert decisions[0].decision.dry_run is True
|
|
|
|
|
|
def test_handoff_matrix_detects_conflict_and_rollback() -> None:
|
|
matrix = HandoffMatrix()
|
|
profile = ControlProfile(
|
|
actuator_entity_id="light.storage",
|
|
handoff_mode=HandoffMode.CONTROLLED,
|
|
related_automation_ids=["automation.storage"],
|
|
)
|
|
|
|
assert matrix.classify(profile) is HandoffMode.CONFLICT
|
|
controlled = matrix.assume_control(profile)
|
|
assert controlled.paused_automation_ids == ["automation.storage"]
|
|
rolled_back = matrix.rollback(controlled)
|
|
assert rolled_back.handoff_mode is HandoffMode.ROLLBACK
|
|
assert rolled_back.paused_automation_ids == []
|
|
|
|
|
|
def test_event_core_executes_allowed_active_decision(tmp_path: Path) -> None:
|
|
stores = FutureStores(tmp_path)
|
|
stores.save_runtime(
|
|
RuntimeState(
|
|
entities={
|
|
"light.storage": EntityState(
|
|
entity_id="light.storage",
|
|
domain="light",
|
|
state="off",
|
|
)
|
|
}
|
|
)
|
|
)
|
|
stores.save_learning(
|
|
LearningState(
|
|
profiles={
|
|
"light.storage": LearningProfile(
|
|
actuator_entity_id="light.storage",
|
|
patterns=[
|
|
BehaviorPatternV2(
|
|
actuator_entity_id="light.storage",
|
|
target_state="on",
|
|
trigger_entity_id="binary_sensor.storage_door",
|
|
trigger_state="on",
|
|
support=3,
|
|
confidence=0.95,
|
|
)
|
|
],
|
|
)
|
|
}
|
|
)
|
|
)
|
|
stores.save_control(
|
|
stores.control().model_copy(
|
|
update={
|
|
"profiles": {
|
|
"light.storage": ControlProfile(
|
|
actuator_entity_id="light.storage",
|
|
stage=SafetyStage.ACTIVE,
|
|
min_confidence=0.8,
|
|
)
|
|
}
|
|
}
|
|
)
|
|
)
|
|
calls: list[str] = []
|
|
|
|
def execute(decision: Decision) -> bool:
|
|
calls.append(decision.actuator_entity_id)
|
|
return True
|
|
|
|
audit = EventCore(stores).process_state_event(
|
|
StateEvent(entity_id="binary_sensor.storage_door", new_state="on"),
|
|
execute=execute,
|
|
)
|
|
|
|
decision = [item.decision for item in audit if item.decision is not None][0]
|
|
assert calls == ["light.storage"]
|
|
assert decision.executed is True
|