alexei.dolgolyov
492bdb95e3
Receive real-time events from games (CS2, Dota 2, LoL, etc.) and drive LED effects through the existing color strip and value source pipelines. Core: - GameEventBus (thread-safe pub/sub) with standardized 23-type event vocabulary - GameAdapter ABC + AdapterRegistry + MappingAdapter (YAML-driven) - Built-in adapters: CS2 GSI, Dota 2 GSI, LoL Live Client, Generic Webhook - Community YAML adapters: Minecraft, Valorant, Rocket League - GameEventColorStripStream with 5 effects (flash/pulse/sweep/color_shift/breathing) - GameEventValueSource with EMA smoothing and timeout - 4 built-in effect presets (FPS Combat, MOBA Health, Racing, Generic Alert) - Auto-setup for Valve GSI games (Steam path detection, cfg file writing) - Demo capture engine exposed to non-demo mode Frontend: - Game tab in Streams tree navigation with integration cards - Game integration editor modal with adapter picker, config fields, event mappings - game_event source type in CSS and ValueSource editors - Setup instructions overlay (markdown rendered) - Live event monitor and connection test API: - Full CRUD for game integrations - Event ingestion endpoint (adapter-level auth) - Adapter metadata, presets, auto-setup, status/diagnostics endpoints
287 lines
9.8 KiB
Python
287 lines
9.8 KiB
Python
"""Tests for AutomationStore — CRUD, rules, name uniqueness."""
|
|
|
|
import pytest
|
|
|
|
from wled_controller.storage.automation import (
|
|
ApplicationRule,
|
|
Automation,
|
|
DisplayStateRule,
|
|
MQTTRule,
|
|
Rule,
|
|
StartupRule,
|
|
SystemIdleRule,
|
|
TimeOfDayRule,
|
|
WebhookRule,
|
|
)
|
|
from wled_controller.storage.automation_store import AutomationStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_db) -> AutomationStore:
|
|
return AutomationStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRuleModels:
|
|
def test_application_round_trip(self):
|
|
r = ApplicationRule(apps=["chrome.exe", "firefox.exe"], match_type="topmost")
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, ApplicationRule)
|
|
assert restored.apps == ["chrome.exe", "firefox.exe"]
|
|
assert restored.match_type == "topmost"
|
|
|
|
def test_time_of_day_round_trip(self):
|
|
r = TimeOfDayRule(start_time="22:00", end_time="06:00")
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, TimeOfDayRule)
|
|
assert restored.start_time == "22:00"
|
|
assert restored.end_time == "06:00"
|
|
|
|
def test_system_idle_round_trip(self):
|
|
r = SystemIdleRule(idle_minutes=10, when_idle=False)
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, SystemIdleRule)
|
|
assert restored.idle_minutes == 10
|
|
assert restored.when_idle is False
|
|
|
|
def test_display_state_round_trip(self):
|
|
r = DisplayStateRule(state="off")
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, DisplayStateRule)
|
|
assert restored.state == "off"
|
|
|
|
def test_mqtt_round_trip(self):
|
|
r = MQTTRule(topic="home/tv", payload="on", match_mode="contains")
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, MQTTRule)
|
|
assert restored.topic == "home/tv"
|
|
assert restored.match_mode == "contains"
|
|
|
|
def test_webhook_round_trip(self):
|
|
r = WebhookRule(token="abc123")
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, WebhookRule)
|
|
assert restored.token == "abc123"
|
|
|
|
def test_startup_round_trip(self):
|
|
r = StartupRule()
|
|
data = r.to_dict()
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, StartupRule)
|
|
|
|
def test_unknown_rule_type_raises(self):
|
|
with pytest.raises(ValueError, match="Unknown rule type"):
|
|
Rule.from_dict({"rule_type": "nonexistent"})
|
|
|
|
def test_legacy_condition_type_migration(self):
|
|
"""Legacy data with condition_type should still deserialize."""
|
|
data = {"condition_type": "startup"}
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, StartupRule)
|
|
|
|
def test_legacy_always_maps_to_startup(self):
|
|
"""Legacy 'always' condition_type should map to StartupRule."""
|
|
data = {"condition_type": "always"}
|
|
restored = Rule.from_dict(data)
|
|
assert isinstance(restored, StartupRule)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Automation model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutomationModel:
|
|
def test_round_trip(self, make_automation):
|
|
auto = make_automation(
|
|
name="Test Auto",
|
|
rules=[StartupRule(), WebhookRule(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.rules) == 2
|
|
assert isinstance(restored.rules[0], StartupRule)
|
|
assert isinstance(restored.rules[1], WebhookRule)
|
|
assert restored.scene_preset_id == "sp_123"
|
|
assert restored.deactivation_mode == "revert"
|
|
|
|
def test_from_dict_skips_unknown_rules(self):
|
|
data = {
|
|
"id": "a1",
|
|
"name": "Skip",
|
|
"enabled": True,
|
|
"rule_logic": "or",
|
|
"rules": [
|
|
{"rule_type": "startup"},
|
|
{"rule_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.rules) == 1 # unknown was skipped
|
|
|
|
def test_legacy_conditions_field_migration(self):
|
|
"""Legacy data with 'conditions' and 'condition_logic' should migrate."""
|
|
data = {
|
|
"id": "a2",
|
|
"name": "Legacy",
|
|
"enabled": True,
|
|
"condition_logic": "and",
|
|
"conditions": [
|
|
{"condition_type": "startup"},
|
|
{"condition_type": "application", "apps": ["test.exe"], "match_type": "running"},
|
|
],
|
|
"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 auto.rule_logic == "and"
|
|
assert len(auto.rules) == 2
|
|
assert isinstance(auto.rules[0], StartupRule)
|
|
assert isinstance(auto.rules[1], ApplicationRule)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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.rule_logic == "or"
|
|
assert store.count() == 1
|
|
|
|
def test_create_with_rules(self, store):
|
|
rules = [
|
|
StartupRule(),
|
|
WebhookRule(token="secret123"),
|
|
]
|
|
a = store.create_automation(
|
|
name="Full",
|
|
enabled=False,
|
|
rule_logic="and",
|
|
rules=rules,
|
|
scene_preset_id="sp_001",
|
|
deactivation_mode="fallback_scene",
|
|
deactivation_scene_preset_id="sp_002",
|
|
tags=["test"],
|
|
)
|
|
assert a.enabled is False
|
|
assert a.rule_logic == "and"
|
|
assert len(a.rules) == 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_rules(self, store):
|
|
a = store.create_automation(name="Rules")
|
|
new_rules = [ApplicationRule(apps=["notepad.exe"])]
|
|
updated = store.update_automation(a.id, rules=new_rules)
|
|
assert len(updated.rules) == 1
|
|
assert isinstance(updated.rules[0], ApplicationRule)
|
|
|
|
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",
|
|
rules=[WebhookRule(token="t1")],
|
|
)
|
|
aid = a.id
|
|
|
|
s2 = AutomationStore(db)
|
|
loaded = s2.get_automation(aid)
|
|
assert loaded.name == "Persist"
|
|
assert len(loaded.rules) == 1
|
|
assert isinstance(loaded.rules[0], WebhookRule)
|
|
db.close()
|