mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
malice
8bc9b14dbaMove wfigs wire rendering into notifications/formatters/fire.py and the
full fire state machine into notifications/gating/fire.py, behind the
registry, NO cutover. handle_wfigs builds a canonical dict, calls
decide(), keeps the inline fires INSERT/UPDATE of current_* and the
tombstoned_at stamp unconditional, then branches on is_cutover(...) —
legacy _attach_commit_handles/all-clear path stays byte-identical while
the new path bakes in shadow.
Reproduces every legacy stamp through GateResult.data_patch:
- forward-only acres/containment growth + 8h cooldown gating
- tombstone wildfire_closed all-clear (row exists AND last_broadcast_at
IS NOT NULL) with _severity_override="priority", _dedup_suffix="closed"
- growth _dedup_suffix=f"{acres}|{contained_pct}", _cooldown_suffix=irwin_id
- idempotent commit UPSERT of fires(last_broadcast_*) + event_log flip
- full _location_anchor fallback chain (geocoder_city -> resolve_anchor
-> landclass -> county -> state) preserved in the formatter
Registered under the three explicit categories (wildfire_declared,
wildfire_incident, wildfire_closed) rather than the `fire` toggle, so the
family-fallback does NOT capture the still-deferred FIRMS categories
(wildfire_hotspot/new_ignition/wildfire_growth); a registration test
asserts those resolve elsewhere.
Native env/fires.py deferred (missing IRWIN/cause/landclass, no tombstone
concept); non-cutover so store._emit_event won't run it.
tier-a: 19 new golden+gate-sequence tests; wfigs handler 23/23 preserved;
suite at 34-failure baseline (1571 passed).
Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
76 lines
3.4 KiB
Python
76 lines
3.4 KiB
Python
"""Phase-0 scaffold tests: formatter registry empty + dispatch correctness.
|
|
|
|
(a) With an empty registry, get_formatter returns None for real categories.
|
|
(b) A registered dummy formatter is called verbatim — multi-line output is
|
|
returned as-is (no Mode-B single-line cap applied).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from meshai.notifications.formatters import FORMATTERS, get_formatter, register
|
|
from meshai.notifications.events import make_event
|
|
from meshai.notifications.renderers.composer import compose_mesh_message
|
|
|
|
|
|
# ── (a) empty registry returns None for known categories ──────────────────────
|
|
|
|
@pytest.mark.parametrize("category", [
|
|
# earthquake_event removed: Phase-1 registers formatters.quake for it.
|
|
# weather_warning/weather_statement removed: Phase-2 registers formatters.nws.
|
|
# road_closure/work_zone/road_incident/traffic_congestion removed: Phase-2 registers formatters.incident.
|
|
# wildfire_declared/wildfire_incident/wildfire_closed removed: Phase-3b registers formatters.fire.
|
|
"wildfire_hotspot", # FIRMS — still deferred (not migrated)
|
|
"battery_critical",
|
|
])
|
|
def test_get_formatter_returns_none_while_registry_empty(category):
|
|
"""Un-migrated categories must still return None from get_formatter."""
|
|
# Only categories not yet migrated to the formatter+gater architecture.
|
|
assert category not in FORMATTERS, (
|
|
f"Category {category!r} should not be in FORMATTERS yet (not migrated)"
|
|
)
|
|
assert get_formatter(category) is None
|
|
|
|
|
|
# ── (b) registered dummy formatter is dispatched verbatim ────────────────────
|
|
|
|
_SYNTHETIC_CATEGORY = "_test_scaffold_dummy_category_phase0"
|
|
_MULTILINE_OUTPUT = "Line one\nLine two\nLine three"
|
|
|
|
|
|
def _dummy_formatter(event, *, now: float, budget: int) -> str:
|
|
"""Returns a fixed multi-line string to prove verbatim passthrough."""
|
|
return _MULTILINE_OUTPUT
|
|
|
|
|
|
def test_registered_formatter_returns_verbatim_multiline(monkeypatch):
|
|
"""Register a dummy for a synthetic category; compose_mesh_message must
|
|
return the dummy's multi-line output verbatim — newlines preserved — not
|
|
re-processed through Mode-B's single-line budget loop.
|
|
|
|
Cutover gate: the formatter is only dispatched when the category appears in
|
|
MESHAI_CUTOVER_CATEGORIES. This test sets the var to verify the formatter
|
|
IS invoked once cut over (the complementary not-cutover case is covered by
|
|
test_cutover_gate.py).
|
|
"""
|
|
from meshai.notifications.cutover import _clear_cache as _cutover_clear
|
|
# Mark synthetic category as cut over for this test.
|
|
monkeypatch.setenv("MESHAI_CUTOVER_CATEGORIES", _SYNTHETIC_CATEGORY)
|
|
_cutover_clear()
|
|
# Register the dummy (clean up afterwards to avoid cross-test pollution).
|
|
register(_SYNTHETIC_CATEGORY, _dummy_formatter)
|
|
try:
|
|
event = make_event(
|
|
source="test",
|
|
category=_SYNTHETIC_CATEGORY,
|
|
severity="routine",
|
|
title="First line\nSecond line", # multi-line title
|
|
)
|
|
result = compose_mesh_message(event)
|
|
assert result == _MULTILINE_OUTPUT, (
|
|
f"Expected verbatim multi-line output, got: {result!r}"
|
|
)
|
|
# Verify newlines are preserved (Mode-B would strip them).
|
|
assert "\n" in result, "Newlines must survive the formatter dispatch path"
|
|
finally:
|
|
FORMATTERS.pop(_SYNTHETIC_CATEGORY, None)
|
|
_cutover_clear() # restore cache for subsequent tests
|