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
8.8 KiB
Python
287 lines
8.8 KiB
Python
"""Tests for AutomationEngine — rule evaluation in isolation."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from wled_controller.core.automations.automation_engine import AutomationEngine
|
|
from wled_controller.storage.automation import (
|
|
ApplicationRule,
|
|
Automation,
|
|
DisplayStateRule,
|
|
StartupRule,
|
|
SystemIdleRule,
|
|
TimeOfDayRule,
|
|
WebhookRule,
|
|
)
|
|
from wled_controller.storage.automation_store import AutomationStore
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_store(tmp_db) -> AutomationStore:
|
|
return AutomationStore(tmp_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_manager():
|
|
m = MagicMock()
|
|
m.fire_event = MagicMock()
|
|
return m
|
|
|
|
|
|
@pytest.fixture
|
|
def engine(mock_store, mock_manager) -> AutomationEngine:
|
|
"""Build an AutomationEngine with the PlatformDetector mocked out.
|
|
|
|
PlatformDetector starts a Windows display-power listener thread that
|
|
causes access violations in the test environment, so we replace it
|
|
with a simple MagicMock.
|
|
"""
|
|
with patch("wled_controller.core.automations.automation_engine.PlatformDetector"):
|
|
eng = AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
)
|
|
return eng
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule evaluation (unit-level)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRuleEvaluation:
|
|
"""Test _evaluate_rule for each rule type individually."""
|
|
|
|
def _make_automation(self, rules):
|
|
now = datetime.now(timezone.utc)
|
|
return Automation(
|
|
id="test_auto",
|
|
name="Test",
|
|
enabled=True,
|
|
rule_logic="or",
|
|
rules=rules,
|
|
scene_preset_id=None,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
def _eval(self, engine, rule, **kwargs):
|
|
"""Invoke the private _evaluate_rule method."""
|
|
defaults = dict(
|
|
running_procs=set(),
|
|
topmost_proc=None,
|
|
topmost_fullscreen=False,
|
|
fullscreen_procs=set(),
|
|
idle_seconds=None,
|
|
display_state=None,
|
|
)
|
|
defaults.update(kwargs)
|
|
return engine._evaluate_rule(
|
|
rule,
|
|
defaults["running_procs"],
|
|
defaults["topmost_proc"],
|
|
defaults["topmost_fullscreen"],
|
|
defaults["fullscreen_procs"],
|
|
defaults["idle_seconds"],
|
|
defaults["display_state"],
|
|
)
|
|
|
|
def test_startup_true(self, engine):
|
|
assert self._eval(engine, StartupRule()) is True
|
|
|
|
def test_application_running_match(self, engine):
|
|
rule = ApplicationRule(apps=["chrome.exe"], match_type="running")
|
|
result = self._eval(
|
|
engine,
|
|
rule,
|
|
running_procs={"chrome.exe", "explorer.exe"},
|
|
)
|
|
assert result is True
|
|
|
|
def test_application_running_no_match(self, engine):
|
|
rule = ApplicationRule(apps=["chrome.exe"], match_type="running")
|
|
result = self._eval(
|
|
engine,
|
|
rule,
|
|
running_procs={"explorer.exe"},
|
|
)
|
|
assert result is False
|
|
|
|
def test_application_topmost_match(self, engine):
|
|
rule = ApplicationRule(apps=["game.exe"], match_type="topmost")
|
|
result = self._eval(
|
|
engine,
|
|
rule,
|
|
topmost_proc="game.exe",
|
|
)
|
|
assert result is True
|
|
|
|
def test_application_topmost_no_match(self, engine):
|
|
rule = ApplicationRule(apps=["game.exe"], match_type="topmost")
|
|
result = self._eval(
|
|
engine,
|
|
rule,
|
|
topmost_proc="chrome.exe",
|
|
)
|
|
assert result is False
|
|
|
|
def test_time_of_day_within_range(self, engine):
|
|
rule = TimeOfDayRule(start_time="00:00", end_time="23:59")
|
|
result = self._eval(engine, rule)
|
|
assert result is True
|
|
|
|
def test_system_idle_when_idle(self, engine):
|
|
rule = SystemIdleRule(idle_minutes=5, when_idle=True)
|
|
result = self._eval(engine, rule, idle_seconds=600.0) # 10 minutes idle
|
|
assert result is True
|
|
|
|
def test_system_idle_not_idle(self, engine):
|
|
rule = SystemIdleRule(idle_minutes=5, when_idle=True)
|
|
result = self._eval(engine, rule, idle_seconds=60.0) # 1 minute idle
|
|
assert result is False
|
|
|
|
def test_system_idle_when_not_idle(self, engine):
|
|
"""when_idle=False means active when user is NOT idle."""
|
|
rule = SystemIdleRule(idle_minutes=5, when_idle=False)
|
|
result = self._eval(engine, rule, idle_seconds=60.0) # 1 min idle (not yet 5)
|
|
assert result is True
|
|
|
|
def test_display_state_match(self, engine):
|
|
rule = DisplayStateRule(state="on")
|
|
result = self._eval(engine, rule, display_state="on")
|
|
assert result is True
|
|
|
|
def test_display_state_no_match(self, engine):
|
|
rule = DisplayStateRule(state="off")
|
|
result = self._eval(engine, rule, display_state="on")
|
|
assert result is False
|
|
|
|
def test_webhook_active(self, engine):
|
|
rule = WebhookRule(token="tok123")
|
|
engine._webhook_states["tok123"] = True
|
|
result = self._eval(engine, rule)
|
|
assert result is True
|
|
|
|
def test_webhook_inactive(self, engine):
|
|
rule = WebhookRule(token="tok123")
|
|
# Not in _webhook_states → False
|
|
result = self._eval(engine, rule)
|
|
assert result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule logic (AND / OR)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRuleLogic:
|
|
def _make_automation(self, rules, logic="or"):
|
|
now = datetime.now(timezone.utc)
|
|
return Automation(
|
|
id="logic_auto",
|
|
name="Logic",
|
|
enabled=True,
|
|
rule_logic=logic,
|
|
rules=rules,
|
|
scene_preset_id=None,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
def test_or_any_true(self, engine):
|
|
auto = self._make_automation(
|
|
[
|
|
ApplicationRule(apps=["missing.exe"], match_type="running"),
|
|
StartupRule(),
|
|
],
|
|
logic="or",
|
|
)
|
|
result = engine._evaluate_rules(
|
|
auto,
|
|
running_procs=set(),
|
|
topmost_proc=None,
|
|
topmost_fullscreen=False,
|
|
fullscreen_procs=set(),
|
|
idle_seconds=None,
|
|
display_state=None,
|
|
)
|
|
assert result is True
|
|
|
|
def test_and_all_must_be_true(self, engine):
|
|
auto = self._make_automation(
|
|
[
|
|
StartupRule(),
|
|
ApplicationRule(apps=["missing.exe"], match_type="running"),
|
|
],
|
|
logic="and",
|
|
)
|
|
result = engine._evaluate_rules(
|
|
auto,
|
|
running_procs=set(),
|
|
topmost_proc=None,
|
|
topmost_fullscreen=False,
|
|
fullscreen_procs=set(),
|
|
idle_seconds=None,
|
|
display_state=None,
|
|
)
|
|
assert result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Webhook state management
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWebhookState:
|
|
@pytest.mark.asyncio
|
|
async def test_set_webhook_state_activate(self, engine):
|
|
await engine.set_webhook_state("tok_1", True)
|
|
assert engine._webhook_states["tok_1"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_webhook_state_deactivate(self, engine):
|
|
engine._webhook_states["tok_1"] = True
|
|
await engine.set_webhook_state("tok_1", False)
|
|
assert engine._webhook_states["tok_1"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Start / Stop lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEngineLifecycle:
|
|
@pytest.mark.asyncio
|
|
async def test_start_creates_task(self, engine):
|
|
await engine.start()
|
|
assert engine._task is not None
|
|
await engine.stop()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_cancels_task(self, engine):
|
|
await engine.start()
|
|
await engine.stop()
|
|
assert engine._task is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_double_start_is_safe(self, engine):
|
|
await engine.start()
|
|
await engine.start() # no-op
|
|
await engine.stop()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_without_start_is_safe(self, engine):
|
|
await engine.stop() # no-op
|