alexei.dolgolyov
2da5c047f9
Some checks failed
Lint & Test / test (push) Failing after 1m33s
All store tests were passing file paths instead of Database objects after the JSON-to-SQLite migration. Updated fixtures to create temp Database instances, rewrote backup e2e tests for binary .db format, and fixed config tests for the simplified StorageConfig.
258 lines
9.0 KiB
Python
258 lines
9.0 KiB
Python
"""Tests for AutomationStore — CRUD, conditions, name uniqueness."""
|
|
|
|
import pytest
|
|
|
|
from wled_controller.storage.automation import (
|
|
AlwaysCondition,
|
|
ApplicationCondition,
|
|
Automation,
|
|
Condition,
|
|
DisplayStateCondition,
|
|
MQTTCondition,
|
|
StartupCondition,
|
|
SystemIdleCondition,
|
|
TimeOfDayCondition,
|
|
WebhookCondition,
|
|
)
|
|
from wled_controller.storage.automation_store import AutomationStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_db) -> AutomationStore:
|
|
return AutomationStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Condition models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestConditionModels:
|
|
def test_always_round_trip(self):
|
|
c = AlwaysCondition()
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, AlwaysCondition)
|
|
|
|
def test_application_round_trip(self):
|
|
c = ApplicationCondition(apps=["chrome.exe", "firefox.exe"], match_type="topmost")
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, ApplicationCondition)
|
|
assert restored.apps == ["chrome.exe", "firefox.exe"]
|
|
assert restored.match_type == "topmost"
|
|
|
|
def test_time_of_day_round_trip(self):
|
|
c = TimeOfDayCondition(start_time="22:00", end_time="06:00")
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, TimeOfDayCondition)
|
|
assert restored.start_time == "22:00"
|
|
assert restored.end_time == "06:00"
|
|
|
|
def test_system_idle_round_trip(self):
|
|
c = SystemIdleCondition(idle_minutes=10, when_idle=False)
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, SystemIdleCondition)
|
|
assert restored.idle_minutes == 10
|
|
assert restored.when_idle is False
|
|
|
|
def test_display_state_round_trip(self):
|
|
c = DisplayStateCondition(state="off")
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, DisplayStateCondition)
|
|
assert restored.state == "off"
|
|
|
|
def test_mqtt_round_trip(self):
|
|
c = MQTTCondition(topic="home/tv", payload="on", match_mode="contains")
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, MQTTCondition)
|
|
assert restored.topic == "home/tv"
|
|
assert restored.match_mode == "contains"
|
|
|
|
def test_webhook_round_trip(self):
|
|
c = WebhookCondition(token="abc123")
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, WebhookCondition)
|
|
assert restored.token == "abc123"
|
|
|
|
def test_startup_round_trip(self):
|
|
c = StartupCondition()
|
|
data = c.to_dict()
|
|
restored = Condition.from_dict(data)
|
|
assert isinstance(restored, StartupCondition)
|
|
|
|
def test_unknown_condition_type_raises(self):
|
|
with pytest.raises(ValueError, match="Unknown condition type"):
|
|
Condition.from_dict({"condition_type": "nonexistent"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Automation model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutomationModel:
|
|
def test_round_trip(self, make_automation):
|
|
auto = make_automation(
|
|
name="Test Auto",
|
|
conditions=[AlwaysCondition(), WebhookCondition(token="tok1")],
|
|
scene_preset_id="sp_123",
|
|
deactivation_mode="revert",
|
|
)
|
|
data = auto.to_dict()
|
|
restored = Automation.from_dict(data)
|
|
|
|
assert restored.id == auto.id
|
|
assert restored.name == "Test Auto"
|
|
assert len(restored.conditions) == 2
|
|
assert isinstance(restored.conditions[0], AlwaysCondition)
|
|
assert isinstance(restored.conditions[1], WebhookCondition)
|
|
assert restored.scene_preset_id == "sp_123"
|
|
assert restored.deactivation_mode == "revert"
|
|
|
|
def test_from_dict_skips_unknown_conditions(self):
|
|
data = {
|
|
"id": "a1",
|
|
"name": "Skip",
|
|
"enabled": True,
|
|
"condition_logic": "or",
|
|
"conditions": [
|
|
{"condition_type": "always"},
|
|
{"condition_type": "future_unknown"},
|
|
],
|
|
"scene_preset_id": None,
|
|
"deactivation_mode": "none",
|
|
"deactivation_scene_preset_id": None,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
auto = Automation.from_dict(data)
|
|
assert len(auto.conditions) == 1 # unknown was skipped
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AutomationStore CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutomationStoreCRUD:
|
|
def test_create(self, store):
|
|
a = store.create_automation(name="Auto A")
|
|
assert a.id.startswith("auto_")
|
|
assert a.name == "Auto A"
|
|
assert a.enabled is True
|
|
assert a.condition_logic == "or"
|
|
assert store.count() == 1
|
|
|
|
def test_create_with_conditions(self, store):
|
|
conditions = [
|
|
AlwaysCondition(),
|
|
WebhookCondition(token="secret123"),
|
|
]
|
|
a = store.create_automation(
|
|
name="Full",
|
|
enabled=False,
|
|
condition_logic="and",
|
|
conditions=conditions,
|
|
scene_preset_id="sp_001",
|
|
deactivation_mode="fallback_scene",
|
|
deactivation_scene_preset_id="sp_002",
|
|
tags=["test"],
|
|
)
|
|
assert a.enabled is False
|
|
assert a.condition_logic == "and"
|
|
assert len(a.conditions) == 2
|
|
assert a.scene_preset_id == "sp_001"
|
|
assert a.tags == ["test"]
|
|
|
|
def test_get_all(self, store):
|
|
store.create_automation("A")
|
|
store.create_automation("B")
|
|
assert len(store.get_all_automations()) == 2
|
|
|
|
def test_get(self, store):
|
|
created = store.create_automation("Get")
|
|
got = store.get_automation(created.id)
|
|
assert got.name == "Get"
|
|
|
|
def test_delete(self, store):
|
|
a = store.create_automation("Del")
|
|
store.delete_automation(a.id)
|
|
assert store.count() == 0
|
|
|
|
def test_delete_not_found(self, store):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.delete_automation("nope")
|
|
|
|
def test_update(self, store):
|
|
a = store.create_automation(name="Old", enabled=True)
|
|
updated = store.update_automation(a.id, name="New", enabled=False)
|
|
assert updated.name == "New"
|
|
assert updated.enabled is False
|
|
|
|
def test_update_conditions(self, store):
|
|
a = store.create_automation(name="Conds")
|
|
new_conds = [ApplicationCondition(apps=["notepad.exe"])]
|
|
updated = store.update_automation(a.id, conditions=new_conds)
|
|
assert len(updated.conditions) == 1
|
|
assert isinstance(updated.conditions[0], ApplicationCondition)
|
|
|
|
def test_update_scene_preset_id_clear(self, store):
|
|
a = store.create_automation(name="SP", scene_preset_id="sp_1")
|
|
updated = store.update_automation(a.id, scene_preset_id="")
|
|
assert updated.scene_preset_id is None
|
|
|
|
def test_update_partial(self, store):
|
|
a = store.create_automation(name="Partial", enabled=True, tags=["orig"])
|
|
updated = store.update_automation(a.id, tags=["new"])
|
|
assert updated.name == "Partial"
|
|
assert updated.enabled is True
|
|
assert updated.tags == ["new"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name uniqueness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutomationNameUniqueness:
|
|
def test_duplicate_name_create(self, store):
|
|
store.create_automation("Dup")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.create_automation("Dup")
|
|
|
|
def test_duplicate_name_update(self, store):
|
|
store.create_automation("First")
|
|
a2 = store.create_automation("Second")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.update_automation(a2.id, name="First")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutomationPersistence:
|
|
def test_persist_and_reload(self, tmp_path):
|
|
from wled_controller.storage.database import Database
|
|
db = Database(tmp_path / "auto_persist.db")
|
|
s1 = AutomationStore(db)
|
|
a = s1.create_automation(
|
|
name="Persist",
|
|
conditions=[WebhookCondition(token="t1")],
|
|
)
|
|
aid = a.id
|
|
|
|
s2 = AutomationStore(db)
|
|
loaded = s2.get_automation(aid)
|
|
assert loaded.name == "Persist"
|
|
assert len(loaded.conditions) == 1
|
|
assert isinstance(loaded.conditions[0], WebhookCondition)
|
|
db.close()
|