781 lines
25 KiB
Python
781 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.actuators.models import (
|
|
BehaviorMode,
|
|
BehaviorPattern,
|
|
BehaviorPrediction,
|
|
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_feedback_marks_prediction_correct_as_learning_pattern(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
|
|
settings = _settings(tmp_path)
|
|
store = ActuatorStore(settings.actuator_store)
|
|
record = store.configure("light.office")
|
|
record = record.model_copy(
|
|
update={
|
|
"assignment": record.assignment.model_copy(
|
|
update={
|
|
"selected_context_entity_ids": [
|
|
"binary_sensor.office_presence"
|
|
],
|
|
}
|
|
),
|
|
"behavior": record.behavior.model_copy(
|
|
update={
|
|
"prediction": BehaviorPrediction(
|
|
target_state="on",
|
|
confidence=0.9,
|
|
generated_at=now,
|
|
reason="test",
|
|
)
|
|
}
|
|
),
|
|
}
|
|
)
|
|
store.upsert(record)
|
|
reader = FakeBehaviorReader(
|
|
entities=[
|
|
HaEntitySummary(entity_id="light.office", domain="light", state="off"),
|
|
HaEntitySummary(
|
|
entity_id="binary_sensor.office_presence",
|
|
domain="binary_sensor",
|
|
state="on",
|
|
),
|
|
],
|
|
history=[],
|
|
logbook=[],
|
|
)
|
|
engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings)
|
|
|
|
result = engine.record_feedback("light.office", correct=True)
|
|
|
|
assert result.behavior.patterns[-1].target_state == "on"
|
|
assert result.behavior.patterns[-1].context_states == {
|
|
"binary_sensor.office_presence": "on"
|
|
}
|
|
assert result.behavior.patterns[-1].source == "user_feedback"
|
|
assert result.behavior.reason == "Vorhersage wurde vom Nutzer als korrekt bestätigt."
|
|
|
|
|
|
def test_feedback_marks_prediction_wrong_and_adds_correction(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
|
|
settings = _settings(tmp_path)
|
|
store = ActuatorStore(settings.actuator_store)
|
|
record = store.configure("light.office")
|
|
record = record.model_copy(
|
|
update={
|
|
"assignment": record.assignment.model_copy(
|
|
update={
|
|
"selected_context_entity_ids": [
|
|
"binary_sensor.office_presence"
|
|
],
|
|
}
|
|
),
|
|
"behavior": record.behavior.model_copy(
|
|
update={
|
|
"patterns": [
|
|
BehaviorPattern(
|
|
target_state="on",
|
|
minute_of_day=60,
|
|
weekday=0,
|
|
context_states={"binary_sensor.office_presence": "on"},
|
|
source="automation",
|
|
weight=1.0,
|
|
observed_at=now - timedelta(days=1),
|
|
)
|
|
],
|
|
"prediction": BehaviorPrediction(
|
|
target_state="on",
|
|
confidence=0.9,
|
|
generated_at=now,
|
|
reason="test",
|
|
),
|
|
}
|
|
),
|
|
}
|
|
)
|
|
store.upsert(record)
|
|
reader = FakeBehaviorReader(
|
|
entities=[
|
|
HaEntitySummary(entity_id="light.office", domain="light", state="off"),
|
|
HaEntitySummary(
|
|
entity_id="binary_sensor.office_presence",
|
|
domain="binary_sensor",
|
|
state="on",
|
|
),
|
|
],
|
|
history=[],
|
|
logbook=[],
|
|
)
|
|
engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings)
|
|
|
|
result = engine.record_feedback(
|
|
"light.office",
|
|
correct=False,
|
|
expected_state="off",
|
|
)
|
|
|
|
assert result.behavior.patterns[0].weight == 0.1
|
|
assert result.behavior.patterns[-1].target_state == "off"
|
|
assert result.behavior.patterns[-1].source == "user_correction"
|
|
assert result.behavior.reason == "Vorhersage wurde vom Nutzer als falsch markiert."
|
|
|
|
|
|
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"),
|
|
("media_player", "off", "turn_off"),
|
|
("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"})
|
|
]
|
|
|
|
|
|
def test_state_change_uses_event_cache_without_rest_state_query(
|
|
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=[],
|
|
history=[],
|
|
logbook=[],
|
|
)
|
|
|
|
def fail_read_entities() -> list[HaEntitySummary]:
|
|
raise AssertionError("Event-Auswertung darf keinen REST-State lesen.")
|
|
|
|
reader.read_entities = fail_read_entities # type: ignore[method-assign]
|
|
engine = BehaviorEngine(ha_reader=reader, store=store, settings=settings)
|
|
|
|
engine.handle_state_change(
|
|
"binary_sensor.storage_door",
|
|
{"state": "on", "last_changed": now.isoformat()},
|
|
current_entities=[
|
|
HaEntitySummary(entity_id="light.storage", domain="light", state="off"),
|
|
HaEntitySummary(
|
|
entity_id="binary_sensor.storage_door",
|
|
domain="binary_sensor",
|
|
state="on",
|
|
last_changed=now,
|
|
),
|
|
],
|
|
)
|
|
|
|
assert reader.service_calls == [
|
|
("light", "turn_on", {"entity_id": "light.storage"})
|
|
]
|