alexei.dolgolyov
6745e25b20
Eight roadmap features from the 2026-06-19 review, each a full vertical (backend + tests + frontend + i18n en/ru/zh); ~67 new unit tests: - automations: SolarRule sunrise/sunset trigger (new utils/solar.py, shared with the daylight cycle; window logic mirrors TimeOfDayRule) - ci: best-effort arm64 multi-arch Docker manifest via QEMU + docker manifest (release.yml; amd64 path untouched, continue-on-error) - game-integration: wire the orphaned LoLPoller via a LoLPollManager + a shared runtime_state module (poll lifecycle on enable/CRUD/startup/shutdown) - ui: color-harmony gradient generator (complementary/analogous/triadic/...) - effects: audio-reactive palette modulation (new audio_energy_tap; brightness/ saturation modulation across all 12 procedural effects) - capture: linear-light blending + spatio-temporal dithering, opt-in per calibration (new utils/linear_light.py, utils/dither.py) - devices: Nanoleaf extControl v2 per-panel UDP streaming (per_panel mode) Also bundles the pending 2026-06-18 production-review fixes and other in-progress work already in the working tree (manual-trigger rule, etc.), since they share files and could not be cleanly separated. Gate: ruff + tsc clean; pytest 2654 passed / 2 skipped. The single failing test (automation manual_trigger handler coverage) is a separate in-progress item owned elsewhere, intentionally left as-is.
692 lines
25 KiB
Python
692 lines
25 KiB
Python
"""Tests for AutomationEngine — rule evaluation in isolation."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from ledgrab.core.automations.automation_engine import AutomationEngine
|
|
from ledgrab.storage.automation import (
|
|
ApplicationRule,
|
|
Automation,
|
|
DisplayStateRule,
|
|
ManualTriggerRule,
|
|
StartupRule,
|
|
SystemIdleRule,
|
|
TimeOfDayRule,
|
|
WebhookRule,
|
|
)
|
|
from ledgrab.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("ledgrab.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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP poll rule — operator helpers + integration with mocked value stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestApplyOperator:
|
|
"""Unit tests for the rule-comparison operator dispatch."""
|
|
|
|
def test_operator_equals(self):
|
|
from ledgrab.core.automations.automation_engine import _apply_operator
|
|
|
|
assert _apply_operator("equals", "ok", "ok") is True
|
|
assert _apply_operator("equals", 42, "42") is True # str coercion
|
|
assert _apply_operator("equals", "ok", "fail") is False
|
|
|
|
def test_operator_contains(self):
|
|
from ledgrab.core.automations.automation_engine import _apply_operator
|
|
|
|
assert _apply_operator("contains", "hello world", "world") is True
|
|
assert _apply_operator("contains", "hello", "xyz") is False
|
|
|
|
def test_operator_regex(self):
|
|
from ledgrab.core.automations.automation_engine import _apply_operator
|
|
|
|
assert _apply_operator("regex", "abc123", r"\d+") is True
|
|
assert _apply_operator("regex", "abc", r"\d+") is False
|
|
assert _apply_operator("regex", "abc", "[") is False # invalid regex
|
|
|
|
def test_operator_gt_lt(self):
|
|
from ledgrab.core.automations.automation_engine import _apply_operator
|
|
|
|
assert _apply_operator("gt", 5, "3") is True
|
|
assert _apply_operator("gt", 2, "3") is False
|
|
assert _apply_operator("lt", 2, "3") is True
|
|
assert _apply_operator("gt", "not-a-number", "3") is False
|
|
|
|
def test_operator_not_equals(self):
|
|
from ledgrab.core.automations.automation_engine import _apply_operator
|
|
|
|
assert _apply_operator("not_equals", "a", "b") is True
|
|
assert _apply_operator("not_equals", "a", "a") is False
|
|
|
|
|
|
class TestHTTPPollRuleEvaluation:
|
|
"""_evaluate_http_poll behavior with a mocked ValueStreamManager.
|
|
|
|
Rule now points at an HTTPValueSource via ``value_source_id``; the
|
|
engine reads the cached raw value from the corresponding ValueStream
|
|
and compares with ``_apply_operator``.
|
|
"""
|
|
|
|
def _engine_with_vsm(self, mock_store, mock_manager, vsm):
|
|
with patch("ledgrab.core.automations.automation_engine.PlatformDetector"):
|
|
return AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
value_stream_manager=vsm,
|
|
)
|
|
|
|
def _vsm_with_stream(self, vs_id: str, raw_value):
|
|
"""Build a mock ValueStreamManager whose ``peek(vs_id)`` returns a
|
|
stream whose ``get_raw_value()`` yields *raw_value*."""
|
|
vsm = MagicMock()
|
|
stream = MagicMock()
|
|
stream.get_raw_value.return_value = raw_value
|
|
vsm.peek.side_effect = lambda q: stream if q == vs_id else None
|
|
return vsm
|
|
|
|
def test_no_manager_returns_false(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
with patch("ledgrab.core.automations.automation_engine.PlatformDetector"):
|
|
eng = AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok")
|
|
assert eng._evaluate_http_poll(rule) is False
|
|
|
|
def test_no_value_source_id_returns_false(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = MagicMock()
|
|
vsm.peek.return_value = None
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="", operator="equals", value="ok")
|
|
assert eng._evaluate_http_poll(rule) is False
|
|
|
|
def test_stream_missing_returns_false(self, mock_store, mock_manager):
|
|
"""If the value stream hasn't been acquired yet, the rule is False."""
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = MagicMock()
|
|
vsm.peek.return_value = None # nothing acquired
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok")
|
|
assert eng._evaluate_http_poll(rule) is False
|
|
|
|
def test_raw_value_none_returns_false(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = self._vsm_with_stream("vs_x", None)
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok")
|
|
assert eng._evaluate_http_poll(rule) is False
|
|
|
|
def test_equals_match(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = self._vsm_with_stream("vs_x", "playing")
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="playing")
|
|
assert eng._evaluate_http_poll(rule) is True
|
|
|
|
def test_exists_when_value_present(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = self._vsm_with_stream("vs_x", 1)
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="exists")
|
|
assert eng._evaluate_http_poll(rule) is True
|
|
|
|
def test_exists_when_value_none(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = self._vsm_with_stream("vs_x", None)
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="exists")
|
|
assert eng._evaluate_http_poll(rule) is False
|
|
|
|
def test_gt_numeric(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = self._vsm_with_stream("vs_x", 72.5)
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="gt", value="70")
|
|
assert eng._evaluate_http_poll(rule) is True
|
|
|
|
def test_stream_without_raw_value_attr_returns_false(self, mock_store, mock_manager):
|
|
"""A non-HTTP stream type that has no ``get_raw_value`` short-circuits."""
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
plain_stream = object() # no get_raw_value attribute
|
|
vsm = MagicMock()
|
|
vsm.peek.return_value = plain_stream
|
|
eng = self._engine_with_vsm(mock_store, mock_manager, vsm)
|
|
rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok")
|
|
assert eng._evaluate_http_poll(rule) is False
|
|
|
|
|
|
class TestSyncValueStreamRefs:
|
|
"""Engine acquire/release diffing against the ValueStreamManager."""
|
|
|
|
def test_no_manager_is_noop(self, mock_store, mock_manager):
|
|
with patch("ledgrab.core.automations.automation_engine.PlatformDetector"):
|
|
eng = AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
)
|
|
eng._sync_value_stream_refs() # should not raise
|
|
|
|
def test_acquires_referenced_sources(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = MagicMock()
|
|
|
|
with patch("ledgrab.core.automations.automation_engine.PlatformDetector"):
|
|
eng = AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
value_stream_manager=vsm,
|
|
)
|
|
|
|
mock_store.create_automation(
|
|
name="poll1",
|
|
enabled=True,
|
|
rule_logic="or",
|
|
rules=[
|
|
HTTPPollRule(
|
|
value_source_id="vs_abc",
|
|
operator="equals",
|
|
value="ok",
|
|
),
|
|
],
|
|
scene_preset_id=None,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
)
|
|
|
|
eng._sync_value_stream_refs()
|
|
vsm.acquire.assert_called_once_with("vs_abc")
|
|
assert eng._value_sources_acquired == {"vs_abc"}
|
|
|
|
def test_releases_unreferenced_sources(self, mock_store, mock_manager):
|
|
"""When a rule is removed, the engine releases its formerly-held source."""
|
|
vsm = MagicMock()
|
|
|
|
with patch("ledgrab.core.automations.automation_engine.PlatformDetector"):
|
|
eng = AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
value_stream_manager=vsm,
|
|
)
|
|
|
|
eng._value_sources_acquired = {"vs_old"}
|
|
eng._sync_value_stream_refs() # no rules reference it
|
|
vsm.release.assert_called_once_with("vs_old")
|
|
assert eng._value_sources_acquired == set()
|
|
|
|
def test_disabled_automation_ignored(self, mock_store, mock_manager):
|
|
from ledgrab.storage.automation import HTTPPollRule
|
|
|
|
vsm = MagicMock()
|
|
|
|
with patch("ledgrab.core.automations.automation_engine.PlatformDetector"):
|
|
eng = AutomationEngine(
|
|
automation_store=mock_store,
|
|
processor_manager=mock_manager,
|
|
poll_interval=0.1,
|
|
value_stream_manager=vsm,
|
|
)
|
|
|
|
mock_store.create_automation(
|
|
name="off",
|
|
enabled=False,
|
|
rule_logic="or",
|
|
rules=[
|
|
HTTPPollRule(
|
|
value_source_id="vs_xyz",
|
|
operator="equals",
|
|
value="",
|
|
),
|
|
],
|
|
scene_preset_id=None,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
)
|
|
|
|
eng._sync_value_stream_refs()
|
|
vsm.acquire.assert_not_called()
|
|
|
|
|
|
class TestHTTPValueStreamExtraction:
|
|
"""Unit tests for the dot-path extractor used by HTTPValueStream."""
|
|
|
|
def test_empty_path_returns_body_text(self):
|
|
from ledgrab.core.processing.value_stream import _extract_simple_path
|
|
|
|
assert _extract_simple_path(None, "", "raw body") == "raw body"
|
|
|
|
def test_simple_key(self):
|
|
from ledgrab.core.processing.value_stream import _extract_simple_path
|
|
|
|
assert _extract_simple_path({"status": "ok"}, "status", "") == "ok"
|
|
|
|
def test_nested_with_array_index(self):
|
|
from ledgrab.core.processing.value_stream import _extract_simple_path
|
|
|
|
body = {"MediaContainer": {"size": 2, "Metadata": [{"title": "Show"}]}}
|
|
assert _extract_simple_path(body, "MediaContainer.size", "") == 2
|
|
assert _extract_simple_path(body, "MediaContainer.Metadata[0].title", "") == "Show"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Manual trigger — one-shot apply gated by the automation's rules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestManualTrigger:
|
|
"""fire_manual_trigger: evaluate rules with the manual term True, then
|
|
apply the scene once (without entering the sticky active state)."""
|
|
|
|
def _make(self, rules, *, enabled=True, logic="or", scene=None, aid="auto_manual"):
|
|
now = datetime.now(timezone.utc)
|
|
return Automation(
|
|
id=aid,
|
|
name="Manual",
|
|
enabled=enabled,
|
|
rule_logic=logic,
|
|
rules=rules,
|
|
scene_preset_id=scene,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
def test_handle_manual_inert_by_default(self, engine):
|
|
# Outside a manual fire the rule reads False, so a manual-only
|
|
# automation never activates from the background tick.
|
|
assert engine._handle_manual(ManualTriggerRule(), None) is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fires_no_scene_one_shot(self, engine):
|
|
auto = self._make([ManualTriggerRule()])
|
|
status, errors = await engine.fire_manual_trigger(auto)
|
|
assert status == "triggered"
|
|
assert errors == []
|
|
# One-shot: records last_activated but does NOT enter the sticky state
|
|
# (so the background tick has nothing to reconcile away → no bounce).
|
|
assert auto.id not in engine._active_automations
|
|
assert auto.id in engine._last_activated
|
|
# The transient flag is always cleared after the evaluation.
|
|
assert engine._manual_fire_active is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skipped_when_and_companion_false(self, engine):
|
|
# manual AND an unset webhook → AND fails → nothing applied.
|
|
auto = self._make([ManualTriggerRule(), WebhookRule(token="nope")], logic="and")
|
|
status, errors = await engine.fire_manual_trigger(auto)
|
|
assert status == "skipped"
|
|
assert errors == []
|
|
assert auto.id not in engine._last_activated
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fires_or_when_companion_false(self, engine):
|
|
# manual OR an unset webhook → manual alone satisfies "or".
|
|
auto = self._make([ManualTriggerRule(), WebhookRule(token="nope")], logic="or")
|
|
status, _errors = await engine.fire_manual_trigger(auto)
|
|
assert status == "triggered"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_works_when_disabled(self, engine):
|
|
# enabled gates only the background loop; a manual trigger ignores it.
|
|
auto = self._make([ManualTriggerRule()], enabled=False)
|
|
status, _errors = await engine.fire_manual_trigger(auto)
|
|
assert status == "triggered"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manual_only_automation_inert_in_background_tick(self, engine, mock_store):
|
|
created = mock_store.create_automation(
|
|
name="manual-bg",
|
|
enabled=True,
|
|
rule_logic="or",
|
|
rules=[ManualTriggerRule()],
|
|
scene_preset_id=None,
|
|
)
|
|
await engine.trigger_evaluate()
|
|
assert created.id not in engine._active_automations
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audits_with_user_actor(self, engine):
|
|
rec = MagicMock()
|
|
with patch("ledgrab.core.activity_log.recorder.get_module_recorder", return_value=rec):
|
|
await engine.fire_manual_trigger(self._make([ManualTriggerRule()]))
|
|
rec.record.assert_called_once()
|
|
kwargs = rec.record.call_args.kwargs
|
|
assert kwargs["action"] == "automation.triggered"
|
|
# No explicit actor → recorder resolves the current user (not "system").
|
|
assert "actor" not in kwargs
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_applies_scene_activated_maps_to_triggered(self, engine):
|
|
engine._scene_preset_store = MagicMock()
|
|
preset = MagicMock()
|
|
preset.name = "Scene"
|
|
engine._scene_preset_store.get_preset.return_value = preset
|
|
engine._target_store = MagicMock()
|
|
engine._device_store = MagicMock()
|
|
auto = self._make([ManualTriggerRule()], scene="scene_x")
|
|
with patch(
|
|
"ledgrab.core.scenes.scene_activator.apply_scene_state",
|
|
new=AsyncMock(return_value=("activated", [])),
|
|
) as apply_mock:
|
|
status, errors = await engine.fire_manual_trigger(auto)
|
|
apply_mock.assert_awaited_once()
|
|
assert status == "triggered"
|
|
assert errors == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_applies_scene_partial_passthrough(self, engine):
|
|
engine._scene_preset_store = MagicMock()
|
|
preset = MagicMock()
|
|
preset.name = "Scene"
|
|
engine._scene_preset_store.get_preset.return_value = preset
|
|
engine._target_store = MagicMock()
|
|
engine._device_store = MagicMock()
|
|
auto = self._make([ManualTriggerRule()], scene="scene_x")
|
|
with patch(
|
|
"ledgrab.core.scenes.scene_activator.apply_scene_state",
|
|
new=AsyncMock(return_value=("partial", ["dev1: timeout"])),
|
|
):
|
|
status, errors = await engine.fire_manual_trigger(auto)
|
|
assert status == "partial"
|
|
assert errors == ["dev1: timeout"]
|
|
|
|
def test_chained_indices(self):
|
|
from ledgrab.core.processing.value_stream import _extract_simple_path
|
|
|
|
body = {"matrix": [[1, 2, 3], [4, 5, 6]]}
|
|
assert _extract_simple_path(body, "matrix[0][1]", "") == 2
|
|
assert _extract_simple_path(body, "matrix[1][2]", "") == 6
|
|
|
|
def test_missing_returns_none(self):
|
|
from ledgrab.core.processing.value_stream import _extract_simple_path
|
|
|
|
assert _extract_simple_path({"a": 1}, "b", "") is None
|
|
assert _extract_simple_path({"a": [1]}, "a[5]", "") is None
|