mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
malice
f06bf3ce92Foundation for making all hazard formatting+gating source-agnostic. ZERO behavior change — the formatter/decider registries are empty (get_formatter/ get_decider return None → existing precomposed/Mode-B path preserved), and the shadow comparator is off unless MESHAI_SHADOW_CATEGORIES is set. - notifications/formatters/ (registry+dispatch with family fallback), gating/ (GateResult + deferred-commit contract), both empty registries. - notifications/clock.py determinism seam; route wfigs/quake/nws gating time reads through it (identical values) so goldens can freeze time. - formatters/_budget.py = copy of central/budget.py; central/budget.py is now a re-export shim (import-smoke test guards it). - compose_mesh_message consults the registry first (verbatim, no Mode-B re-cap), falls back to legacy; _resolve_budget injects per-category budget. - notifications/shadow.py + two DRY-RUN hooks (consumer._normalize, dispatcher render): compute the new result and diff-log SHADOW_MISMATCH JSONL, but NEVER commit/emit/write tables and always broadcast the OLD result. Inert by default. - tests/harness (pinned_time/pinned_tz, byte-golden, gate-sequence) + scripts/capture_fixtures.py (ephemeral read-only NATS capture); tzdata pinned. Tests: +60 (18 scaffold + 42 harness/shadow); 0 new failures (34 baseline). Co-authored-by: Matt Johnson <mj@k7zvx.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
59 lines
2 KiB
Python
59 lines
2 KiB
Python
"""Phase-0: verify the clock seam is monkeypatchable and handlers use it.
|
|
|
|
Tests:
|
|
1. clock.now() returns a float close to time.time() at runtime.
|
|
2. Monkeypatching clock.now propagates into the three refactored handlers
|
|
(quake_handler._now, nws_handler._now, wfigs_handler._now).
|
|
"""
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
import meshai.notifications.clock as clock_mod
|
|
import meshai.central.quake_handler as quake_handler
|
|
import meshai.central.nws_handler as nws_handler
|
|
import meshai.central.wfigs_handler as wfigs_handler
|
|
|
|
|
|
_FROZEN_TS = 1_700_000_000.0
|
|
|
|
|
|
def test_clock_now_returns_float_close_to_time():
|
|
before = time.time()
|
|
result = clock_mod.now()
|
|
after = time.time()
|
|
assert isinstance(result, float), "clock.now() must return float"
|
|
assert before <= result <= after + 0.1, "clock.now() must be current time"
|
|
|
|
|
|
def test_clock_now_is_monkeypatchable(monkeypatch):
|
|
monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
|
|
assert clock_mod.now() == _FROZEN_TS
|
|
|
|
|
|
def test_quake_handler_now_uses_clock_seam(monkeypatch):
|
|
"""quake_handler._now() must reflect a monkeypatched clock.now."""
|
|
monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
|
|
result = quake_handler._now()
|
|
assert result == int(_FROZEN_TS), (
|
|
f"quake_handler._now() returned {result!r}, expected {int(_FROZEN_TS)}"
|
|
)
|
|
|
|
|
|
def test_nws_handler_now_uses_clock_seam(monkeypatch):
|
|
"""nws_handler._now() must reflect a monkeypatched clock.now."""
|
|
monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
|
|
result = nws_handler._now()
|
|
assert result == int(_FROZEN_TS), (
|
|
f"nws_handler._now() returned {result!r}, expected {int(_FROZEN_TS)}"
|
|
)
|
|
|
|
|
|
def test_wfigs_handler_now_uses_clock_seam(monkeypatch):
|
|
"""wfigs_handler._now() must reflect a monkeypatched clock.now."""
|
|
monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
|
|
result = wfigs_handler._now()
|
|
assert result == int(_FROZEN_TS), (
|
|
f"wfigs_handler._now() returned {result!r}, expected {int(_FROZEN_TS)}"
|
|
)
|