feat(events): processor handles SYSTEM_ARM_REQUEST over MQTT
Adds _handle_arm_request and a dedicated bus.subscribe on
Topics.SYSTEM_ARM_REQUEST. Payload {mode, pin, triggered_by} is
dispatched to ArmStateFSM.transition, which verifies the PIN via
alerts.pin.verify_pin and performs the state change.
This is the missing link for web /system/api/arm to actually move
the system into an armed state. Part of issue #2.
This commit is contained in:
@@ -511,3 +511,54 @@ class TestEventsPublishedBroadcast:
|
|||||||
)
|
)
|
||||||
|
|
||||||
assert not any(t == Topics.EVENTS_PUBLISHED for t, _ in bus.published)
|
assert not any(t == Topics.EVENTS_PUBLISHED for t, _ in bus.published)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Arm Request Dispatch
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestArmRequestDispatch:
|
||||||
|
"""SYSTEM_ARM_REQUEST messages must reach ArmStateFSM.transition."""
|
||||||
|
|
||||||
|
def test_arm_request_calls_fsm_transition(self, test_db):
|
||||||
|
from vigilar.events.processor import EventProcessor
|
||||||
|
|
||||||
|
processor = EventProcessor.__new__(EventProcessor)
|
||||||
|
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
class FakeFSM:
|
||||||
|
state = ArmState.DISARMED
|
||||||
|
|
||||||
|
def transition(self, new_state, pin="", triggered_by="system"):
|
||||||
|
calls.append((new_state, pin, triggered_by))
|
||||||
|
return True
|
||||||
|
|
||||||
|
processor._handle_arm_request(
|
||||||
|
payload={"mode": "ARMED_AWAY", "pin": "1234", "triggered_by": "web"},
|
||||||
|
fsm=FakeFSM(),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert len(calls) == 1
|
||||||
|
new_state, pin, triggered_by = calls[0]
|
||||||
|
assert new_state == ArmState.ARMED_AWAY
|
||||||
|
assert pin == "1234"
|
||||||
|
assert triggered_by == "web"
|
||||||
|
|
||||||
|
def test_arm_request_ignores_bad_mode(self, test_db):
|
||||||
|
from vigilar.events.processor import EventProcessor
|
||||||
|
|
||||||
|
processor = EventProcessor.__new__(EventProcessor)
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
class FakeFSM:
|
||||||
|
def transition(self, *a, **kw):
|
||||||
|
calls.append((a, kw))
|
||||||
|
return True
|
||||||
|
|
||||||
|
processor._handle_arm_request(
|
||||||
|
payload={"mode": "NONSENSE", "pin": "1234"},
|
||||||
|
fsm=FakeFSM(),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert calls == []
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from sqlalchemy.engine import Engine
|
|||||||
|
|
||||||
from vigilar.bus import MessageBus
|
from vigilar.bus import MessageBus
|
||||||
from vigilar.config import VigilarConfig
|
from vigilar.config import VigilarConfig
|
||||||
from vigilar.constants import EventType, Severity, Topics
|
from vigilar.constants import ArmState, EventType, Severity, Topics
|
||||||
from vigilar.events.rules import RuleEngine
|
from vigilar.events.rules import RuleEngine
|
||||||
from vigilar.events.state import ArmStateFSM
|
from vigilar.events.state import ArmStateFSM
|
||||||
from vigilar.storage.db import get_db_path, init_db
|
from vigilar.storage.db import get_db_path, init_db
|
||||||
@@ -58,12 +58,20 @@ class EventProcessor:
|
|||||||
fsm.set_bus(bus)
|
fsm.set_bus(bus)
|
||||||
bus.connect()
|
bus.connect()
|
||||||
|
|
||||||
# Subscribe to all Vigilar topics
|
# Subscribe to all Vigilar topics (events/motion/sensors/etc.)
|
||||||
def on_message(topic: str, payload: dict[str, Any]) -> None:
|
def on_message(topic: str, payload: dict[str, Any]) -> None:
|
||||||
self._handle_event(topic, payload, engine, fsm, rule_engine, bus)
|
self._handle_event(topic, payload, engine, fsm, rule_engine, bus)
|
||||||
|
|
||||||
bus.subscribe_all(on_message)
|
bus.subscribe_all(on_message)
|
||||||
|
|
||||||
|
# Dedicated subscription for web-originated arm/disarm requests.
|
||||||
|
# Kept separate from on_message because these are commands, not
|
||||||
|
# classifiable events.
|
||||||
|
def on_arm_request(topic: str, payload: dict[str, Any]) -> None:
|
||||||
|
self._handle_arm_request(payload, fsm)
|
||||||
|
|
||||||
|
bus.subscribe(Topics.SYSTEM_ARM_REQUEST, on_arm_request)
|
||||||
|
|
||||||
log.info("Event processor started")
|
log.info("Event processor started")
|
||||||
|
|
||||||
# Main loop
|
# Main loop
|
||||||
@@ -160,6 +168,26 @@ class EventProcessor:
|
|||||||
except Exception:
|
except Exception:
|
||||||
log.exception("Error processing event on %s", topic)
|
log.exception("Error processing event on %s", topic)
|
||||||
|
|
||||||
|
def _handle_arm_request(
|
||||||
|
self,
|
||||||
|
payload: dict[str, Any],
|
||||||
|
fsm: "ArmStateFSM",
|
||||||
|
) -> None:
|
||||||
|
"""Handle an arm/disarm request received over MQTT.
|
||||||
|
|
||||||
|
Payload fields:
|
||||||
|
- mode: str — desired ArmState ("DISARMED", "ARMED_HOME", "ARMED_AWAY")
|
||||||
|
- pin: str — plaintext PIN (FSM verifies against security.pin_hash)
|
||||||
|
- triggered_by: str — origin tag for the audit log (e.g. "web")
|
||||||
|
"""
|
||||||
|
mode = payload.get("mode", "")
|
||||||
|
if mode not in ArmState.__members__:
|
||||||
|
log.warning("Ignoring arm request with invalid mode: %r", mode)
|
||||||
|
return
|
||||||
|
pin = payload.get("pin", "")
|
||||||
|
triggered_by = payload.get("triggered_by", "unknown")
|
||||||
|
fsm.transition(ArmState(mode), pin=pin, triggered_by=triggered_by)
|
||||||
|
|
||||||
def _classify_event(
|
def _classify_event(
|
||||||
self, topic: str, payload: dict[str, Any]
|
self, topic: str, payload: dict[str, Any]
|
||||||
) -> tuple[str | None, str | None, str | None]:
|
) -> tuple[str | None, str | None, str | None]:
|
||||||
|
|||||||
Reference in New Issue
Block a user