refactor(automations): rule dispatch via class-level handler table

AutomationEngine._evaluate_rule used to rebuild a 9-entry dispatch
dict on EVERY rule evaluation (audit finding H2). Unknown rule types
silently returned False — adding a new Rule subclass without an entry
just made it inert forever.

Refactor:

  * Per-rule-type bodies are now ``_handle_<kind>(self, rule, ctx)``
    methods on AutomationEngine.
  * A ``_RuleEvalContext`` frozen dataclass bundles all the
    cross-cutting state (running_procs, topmost_proc,
    topmost_fullscreen, fullscreen_procs, idle_seconds, display_state)
    so adding a new handler does not require widening
    ``_evaluate_rule``'s parameter list.
  * ``AutomationEngine._RULE_HANDLERS`` is bound once at module-import
    time after the class is defined.
  * ``_assert_rule_handler_coverage()`` runs at import: every Rule
    subclass imported by the module must have an entry, and entries
    keyed by an unknown class are also rejected.

Unknown-type fallback now logs a warning instead of silently returning
False, so a future Rule subclass missing from the registry surfaces in
operator logs rather than just behaving as if the automation were off.

The pure storage layer (storage/automation.py) is untouched — the
handler bodies stay on the engine where the cross-layer dependencies
(MQTT runtime, HA manager, HTTP endpoint store, webhook state) live.

Tests: 4 new tests cover the rule-type/handler bijection, callable
shape, missing-entry rejection, and unknown-class rejection. 44
existing automation engine tests stay green; ruff clean.
This commit is contained in:
2026-05-22 23:07:07 +03:00
parent 5fec8db901
commit 98fb61d932
2 changed files with 321 additions and 15 deletions
@@ -0,0 +1,80 @@
"""Tests for the AutomationEngine rule-handler dispatch registry.
Lock in the import-time invariant: every Rule subclass imported by the
engine module has a corresponding ``_handle_*`` entry in
``_RULE_HANDLERS``, and the coverage check rejects drift.
"""
from __future__ import annotations
import pytest
from ledgrab.core.automations import automation_engine
from ledgrab.core.automations.automation_engine import (
AutomationEngine,
_assert_rule_handler_coverage,
)
from ledgrab.storage.automation import (
ApplicationRule,
DisplayStateRule,
HomeAssistantRule,
HTTPPollRule,
MQTTRule,
Rule,
StartupRule,
SystemIdleRule,
TimeOfDayRule,
WebhookRule,
)
EXPECTED_RULE_TYPES = {
StartupRule,
ApplicationRule,
TimeOfDayRule,
SystemIdleRule,
DisplayStateRule,
MQTTRule,
WebhookRule,
HomeAssistantRule,
HTTPPollRule,
}
def test_every_rule_type_has_a_handler():
"""The registry exactly covers the rule-type set the engine imports."""
assert set(AutomationEngine._RULE_HANDLERS.keys()) == EXPECTED_RULE_TYPES
def test_handlers_are_engine_methods():
"""Each handler value is a method defined on AutomationEngine."""
for rule_cls, handler in AutomationEngine._RULE_HANDLERS.items():
assert callable(handler), f"handler for {rule_cls.__name__} is not callable"
# Method names start with _handle_
assert handler.__name__.startswith(
"_handle_"
), f"handler for {rule_cls.__name__} has unexpected name {handler.__name__}"
def test_coverage_assertion_raises_when_handler_is_missing(monkeypatch):
"""Removing an entry from _RULE_HANDLERS makes the import-time check fail."""
# Build a clone of the registry without one entry to simulate drift.
original = dict(AutomationEngine._RULE_HANDLERS)
pruned = {k: v for k, v in original.items() if k is not WebhookRule}
monkeypatch.setattr(automation_engine.AutomationEngine, "_RULE_HANDLERS", pruned)
with pytest.raises(RuntimeError, match="WebhookRule"):
_assert_rule_handler_coverage()
def test_coverage_assertion_raises_when_unexpected_handler_added(monkeypatch):
"""An entry keyed by an unknown class is also caught."""
class _UnknownRule(Rule): # type: ignore[misc]
pass
extended = {**AutomationEngine._RULE_HANDLERS, _UnknownRule: lambda *a: True}
monkeypatch.setattr(automation_engine.AutomationEngine, "_RULE_HANDLERS", extended)
with pytest.raises(RuntimeError, match="_UnknownRule"):
_assert_rule_handler_coverage()