feat(automations): expand automation rules + UI + engine coverage

Storage model + Pydantic schema + route surface gain the new rule shapes the engine already supports. Frontend automations editor grows the matching inputs. New core/test_automation_engine.py pins the dispatch table rules behind ~285 lines of unit coverage.
2026-05-23 00:48:19 +03:00
parent f03cb303c3
commit 3fe66d80cb
5 changed files with 533 additions and 3 deletions
@@ -284,3 +284,284 @@ class TestEngineLifecycle:
    @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"
+
+    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