32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
from pigway_frame_flow.models import Power
|
|
|
|
|
|
def test_observer_records_only_changes(service, monkeypatch):
|
|
driver = service.registry.get("living-room")
|
|
baseline = driver.device()
|
|
monkeypatch.setattr(driver, "device", lambda: baseline)
|
|
service.device_observer.tick()
|
|
before = len(service.db.all("SELECT * FROM events"))
|
|
service.device_observer.tick()
|
|
assert len(service.db.all("SELECT * FROM events")) == before
|
|
changed = baseline.model_copy(
|
|
update={"status": "offline", "power": Power(battery_percent=10, charging=False)}
|
|
)
|
|
monkeypatch.setattr(driver, "device", lambda: changed)
|
|
service.device_observer.tick()
|
|
assert {r["event_type"] for r in service.db.all("SELECT * FROM events")} >= {
|
|
"device.offline",
|
|
"device.low_battery",
|
|
}
|
|
|
|
|
|
def test_cloud_failure_is_not_device_offline(service, monkeypatch):
|
|
driver = service.registry.get("living-room")
|
|
|
|
def fail():
|
|
raise RuntimeError("cloud failure")
|
|
|
|
monkeypatch.setattr(driver, "device", fail)
|
|
service.device_observer.tick()
|
|
assert not service.db.all("SELECT * FROM events")
|