diff --git a/work/meshai/central/budget.py b/work/meshai/central/budget.py index e22c978..32f76c7 100644 --- a/work/meshai/central/budget.py +++ b/work/meshai/central/budget.py @@ -1,32 +1,9 @@ -"""Shared per-adapter mesh packet budget helpers. +"""Re-export shim — implementation has moved to meshai.notifications.formatters._budget. -Every broadcast handler fits its final wire string to the live mesh -transport's single-packet character budget. main.py injects the active -transport's `max_chars` (140 for the current LoRa configs) into -adapter_config via set_runtime_override for each broadcast adapter, so -`budget_for(adapter)` returns the runtime value durably across cache -invalidation. Default 140 when no override is present (e.g. unit tests). +Every existing ``from meshai.central.budget import budget_for, fit_to_budget`` +import continues to work unchanged. This shim is the sole consumer of the +canonical implementation; update the implementation there, not here. """ -from __future__ import annotations +from meshai.notifications.formatters._budget import budget_for, fit_to_budget -from meshai.adapter_config import adapter_config - - -def budget_for(adapter: str, default: int = 140) -> int: - """Per-adapter mesh packet budget. Reads adapter_config..single_packet_max_chars, - which main.py overrides at runtime to the live transport max_chars (140). Default 140.""" - try: - return int(getattr(getattr(adapter_config, adapter), "single_packet_max_chars", default)) - except Exception: - return default - - -def fit_to_budget(s: str, limit: int) -> str: - """Trim s to <= limit chars at a word boundary, appending an ellipsis. Never chops a word mid-word.""" - if len(s) <= limit: - return s - cut = s[: max(0, limit - 1)].rstrip() - cut = cut.rsplit(" ", 1)[0] if " " in cut else cut - if not cut: - cut = s[: max(0, limit - 1)] - return cut.rstrip() + "…" +__all__ = ["budget_for", "fit_to_budget"] diff --git a/work/meshai/central/consumer.py b/work/meshai/central/consumer.py index 1ccc150..738c064 100644 --- a/work/meshai/central/consumer.py +++ b/work/meshai/central/consumer.py @@ -590,6 +590,22 @@ class CentralConsumer: # Scheduled broadcasters (band_conditions) bypass _normalize() # entirely -- they enter via Dispatcher.dispatch_scheduled_broadcast() # and are unaffected by this gate. + # Phase-0b shadow gate hook — inert by default. + # Called RIGHT BEFORE the default-deny return so the shadow observes the + # same broadcast decision the legacy handler made (synthesized is not None). + # The OLD result is kept unchanged below; the shadow only observes. + try: + from meshai.notifications.shadow import shadow_gate as _shadow_gate + _shadow_gate( + category, + data, + source=inner.get("adapter") or "central", + now=time.time(), + old_broadcast=synthesized is not None, + ) + except Exception: # noqa: BLE001 — shadow must never affect production + pass + if synthesized is None: logger.debug( "consumer: default-deny -- no handler synthesized for " diff --git a/work/meshai/central/nws_handler.py b/work/meshai/central/nws_handler.py index e571e6a..ad19112 100644 --- a/work/meshai/central/nws_handler.py +++ b/work/meshai/central/nws_handler.py @@ -37,6 +37,8 @@ import zoneinfo from datetime import datetime, timezone from typing import Any, Optional +from meshai.notifications import clock + from meshai.persistence import get_db logger = logging.getLogger(__name__) @@ -130,7 +132,7 @@ def _parse_motion(params: dict) -> tuple: return compass, mph -def _now() -> int: return int(time.time()) +def _now() -> int: return int(clock.now()) def _is_update(conn, d: dict) -> bool: diff --git a/work/meshai/central/quake_handler.py b/work/meshai/central/quake_handler.py index 59b8877..8224b6a 100644 --- a/work/meshai/central/quake_handler.py +++ b/work/meshai/central/quake_handler.py @@ -31,6 +31,8 @@ import math import time from typing import Any, Optional +from meshai.notifications import clock + from meshai.persistence import get_db logger = logging.getLogger(__name__) @@ -41,7 +43,7 @@ logger = logging.getLogger(__name__) # GUI edits take effect on the next envelope without restart. -def _now() -> int: return int(time.time()) +def _now() -> int: return int(clock.now()) def _haversine_mi(lat1, lon1, lat2, lon2) -> float: diff --git a/work/meshai/central/wfigs_handler.py b/work/meshai/central/wfigs_handler.py index 1a96965..b892124 100644 --- a/work/meshai/central/wfigs_handler.py +++ b/work/meshai/central/wfigs_handler.py @@ -35,6 +35,8 @@ import logging import time from typing import Any, Optional +from meshai.notifications import clock + from meshai.persistence import get_db logger = logging.getLogger(__name__) @@ -62,7 +64,7 @@ def _cleanup_stale_fires(conn) -> None: def _now() -> int: - return int(time.time()) + return int(clock.now()) def _fire_too_old_to_announce(declared_at_epoch, now) -> bool: diff --git a/work/meshai/notifications/clock.py b/work/meshai/notifications/clock.py new file mode 100644 index 0000000..45286c0 --- /dev/null +++ b/work/meshai/notifications/clock.py @@ -0,0 +1,23 @@ +"""Determinism seam for time reads across the notification pipeline. + +All handlers and formatters should call clock.now() / clock.now_dt() instead +of time.time() / datetime.now() directly. A single monkeypatch point makes +golden-file tests and time-freeze fixtures trivial: + + monkeypatch.setattr("meshai.notifications.clock.now", lambda: 1_700_000_000.0) + +clock.now() is semantically identical to time.time() at runtime. +""" + +import time +from datetime import datetime + + +def now() -> float: + """Return the current Unix timestamp (float). Monkeypatchable seam.""" + return time.time() + + +def now_dt(tz=None): + """Return the current datetime (optionally timezone-aware). Monkeypatchable seam.""" + return datetime.now(tz) diff --git a/work/meshai/notifications/formatters/__init__.py b/work/meshai/notifications/formatters/__init__.py new file mode 100644 index 0000000..04675b8 --- /dev/null +++ b/work/meshai/notifications/formatters/__init__.py @@ -0,0 +1,55 @@ +"""Formatter registry for the Phase-1+ mesh-message dispatch path. + +FORMATTERS is intentionally empty at this phase (Phase 0 scaffold). +No category is migrated yet — all events fall through to the legacy +compose_mesh_message Mode-B path. Zero behavior change. + +Usage (future phases): + from meshai.notifications.formatters import register + + @register("earthquake_event") + def _fmt_quake(event, *, now: float, budget: int) -> str: + ... +""" + +from typing import Callable, Optional + +# Empty registry — populated by per-category formatter modules (Phase 1+). +FORMATTERS: dict[str, Callable] = {} + + +def register(category: str, fn: Callable) -> Callable: + """Register a formatter callable for a category (or family toggle name). + + Decorates or calls directly: + register("earthquake_event", my_fn) + @register("earthquake_event") + def my_fn(...): ... + """ + FORMATTERS[category] = fn + return fn + + +def get_formatter(category: str) -> Optional[Callable]: + """Return the formatter for *category*, or None if none is registered. + + Resolution order: + 1. Direct category match in FORMATTERS. + 2. Family/toggle fallback: look up the category's toggle name via + get_toggle(), then check FORMATTERS for that toggle key. + 3. None — caller falls through to legacy Mode-B composition. + """ + fn = FORMATTERS.get(category) + if fn is not None: + return fn + # Family fallback — mirrors _category_label()'s toggle lookup in composer.py. + try: + from meshai.notifications.categories import get_toggle + tog = get_toggle(category) + if tog: + fn = FORMATTERS.get(tog) + if fn is not None: + return fn + except Exception: + pass + return None diff --git a/work/meshai/notifications/formatters/_budget.py b/work/meshai/notifications/formatters/_budget.py new file mode 100644 index 0000000..e22c978 --- /dev/null +++ b/work/meshai/notifications/formatters/_budget.py @@ -0,0 +1,32 @@ +"""Shared per-adapter mesh packet budget helpers. + +Every broadcast handler fits its final wire string to the live mesh +transport's single-packet character budget. main.py injects the active +transport's `max_chars` (140 for the current LoRa configs) into +adapter_config via set_runtime_override for each broadcast adapter, so +`budget_for(adapter)` returns the runtime value durably across cache +invalidation. Default 140 when no override is present (e.g. unit tests). +""" +from __future__ import annotations + +from meshai.adapter_config import adapter_config + + +def budget_for(adapter: str, default: int = 140) -> int: + """Per-adapter mesh packet budget. Reads adapter_config..single_packet_max_chars, + which main.py overrides at runtime to the live transport max_chars (140). Default 140.""" + try: + return int(getattr(getattr(adapter_config, adapter), "single_packet_max_chars", default)) + except Exception: + return default + + +def fit_to_budget(s: str, limit: int) -> str: + """Trim s to <= limit chars at a word boundary, appending an ellipsis. Never chops a word mid-word.""" + if len(s) <= limit: + return s + cut = s[: max(0, limit - 1)].rstrip() + cut = cut.rsplit(" ", 1)[0] if " " in cut else cut + if not cut: + cut = s[: max(0, limit - 1)] + return cut.rstrip() + "…" diff --git a/work/meshai/notifications/gating/__init__.py b/work/meshai/notifications/gating/__init__.py new file mode 100644 index 0000000..71eb17f --- /dev/null +++ b/work/meshai/notifications/gating/__init__.py @@ -0,0 +1,50 @@ +"""Gating/budgeting decision registry for the Phase-1+ dispatch path. + +DECIDERS is intentionally empty at this phase (Phase 0 scaffold). +No category is migrated yet — all gating logic remains in the existing +handler modules (wfigs_handler, nws_handler, etc.). Zero behavior change. + +Usage (future phases): + from meshai.notifications.gating import register + from meshai.notifications.gating.base import GateResult + + @register("earthquake_event") + def _gate_quake(event, *, now: float) -> GateResult: + ... +""" + +from typing import Callable, Optional + +# Empty registry — populated by per-category gating modules (Phase 1+). +DECIDERS: dict = {} + + +def register(category: str, fn: Callable) -> Callable: + """Register a gating callable for a category (or family toggle name).""" + DECIDERS[category] = fn + return fn + + +def get_decider(category: str) -> Optional[Callable]: + """Return the gating decider for *category*, or None if none is registered. + + Resolution order: + 1. Direct category match in DECIDERS. + 2. Family/toggle fallback: look up the category's toggle name via + get_toggle(), then check DECIDERS for that toggle key. + 3. None — caller falls through to legacy per-handler gating logic. + """ + fn = DECIDERS.get(category) + if fn is not None: + return fn + # Family fallback — mirrors the toggle lookup pattern in composer.py. + try: + from meshai.notifications.categories import get_toggle + tog = get_toggle(category) + if tog: + fn = DECIDERS.get(tog) + if fn is not None: + return fn + except Exception: + pass + return None diff --git a/work/meshai/notifications/gating/base.py b/work/meshai/notifications/gating/base.py new file mode 100644 index 0000000..b01ad41 --- /dev/null +++ b/work/meshai/notifications/gating/base.py @@ -0,0 +1,49 @@ +"""GateResult — the return type for all gating/budgeting decision functions. + +A decider registered in meshai.notifications.gating.DECIDERS receives an +Event and returns a GateResult. The dispatcher inspects `broadcast` to +decide whether to send, merges `data_patch` into event.data, and calls +`commit(now)` exactly once per channel on confirmed delivery. + +Deferred-commit contract +------------------------ +`commit(now: float)` is called ONCE, ONLY after the message has been +successfully delivered to a mesh channel. Rules: + + * Idempotent: the callback MUST use UPSERTs (INSERT OR REPLACE / ON + CONFLICT DO UPDATE), never bare INSERTs, so a retry on failure does + not create duplicate rows. + + * One call per channel: if the same event is broadcast on N mesh + channels, commit is called N times (once per successful deliver()). + The UPSERT requirement makes that safe. + + * No exceptions may escape commit() — wrap all DB writes in try/except + and log failures; a crashing commit would abort the send loop. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Callable, Optional + + +@dataclass +class GateResult: + """Decision from a gating function. + + Attributes: + broadcast: True → send the event; False → suppress. + lifecycle: Short label for logging/audit (e.g. "new", "update", + "cooldown", "suppress"). + reason: Human-readable explanation used in debug logs. + data_patch: Dict merged into event.data before the message is + rendered and sent (use for attaching callbacks, etc.). + commit: Optional callable invoked once per channel on confirmed + delivery. Signature: commit(now: float) -> None. + See module docstring for the full contract. + """ + broadcast: bool + lifecycle: str = "" + reason: str = "" + data_patch: dict = field(default_factory=dict) + commit: Optional[Callable[[float], None]] = None diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index ba28bba..846cee2 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -443,6 +443,16 @@ class Dispatcher: self._logger.exception("mesh composer crashed; falling back to legacy message") friendly = None + # Phase-0b shadow render hook — inert by default. + # Called AFTER compose_mesh_message so old_wire is the real produced + # string. The real path continues to use *friendly* unchanged below. + if friendly is not None: + try: + from meshai.notifications.shadow import shadow_render as _shadow_render + _shadow_render(event.category, event, old_wire=friendly) + except Exception: # noqa: BLE001 — shadow must never affect production + pass + delivered_any = False for ch_type in ch_types: rule = None diff --git a/work/meshai/notifications/renderers/composer.py b/work/meshai/notifications/renderers/composer.py index 7f204c4..c3c0d01 100644 --- a/work/meshai/notifications/renderers/composer.py +++ b/work/meshai/notifications/renderers/composer.py @@ -20,12 +20,15 @@ exceed the budget, the primary identifier is shrunk by codepoints and suffixed with `…` so the byte budget always holds. """ +import logging import re from html import unescape from typing import Optional from meshai.notifications.events import Event +logger = logging.getLogger(__name__) + # Hard byte budget for a single mesh broadcast line (Matt-approved cap). _BYTE_BUDGET = 150 @@ -294,6 +297,25 @@ def _context_segment(event: Event) -> Optional[str]: return ", ".join(bits) if bits else None +def _resolve_budget(event: Event) -> int: + """Pick a per-adapter packet budget for the new formatter path. + + Mirrors how the existing precomposed handlers select their budget: + wfigs uses budget_for("wfigs"), quake uses budget_for("usgs_quake"), + etc. Here we use event.source (the adapter name) as the budget key. + Falls back to 140 (the universal LoRa default) when source is absent + or budget_for raises. + """ + src = (getattr(event, "source", "") or "").strip() + if src: + try: + from meshai.central.budget import budget_for + return budget_for(src) + except Exception: + pass + return 140 + + def compose_mesh_message(event: Event) -> str: """Compose a friendly mesh-broadcast string with 150-byte UTF-8 hard cap. @@ -306,6 +328,22 @@ def compose_mesh_message(event: Event) -> str: Return it verbatim -- no family-label prefix, no region tail, no severity word append. """ + # Phase-1+ formatter dispatch — EMPTY REGISTRY at Phase 0, so this + # no-ops on every call. When a formatter is registered for the event's + # category (or its toggle family), it is called here and its output + # returned verbatim (newlines preserved, no Mode-B re-entry). + from meshai.notifications.formatters import get_formatter + from meshai.notifications import clock + fmt = get_formatter(event.category) + if fmt is not None: + try: + return fmt(event, now=clock.now(), budget=_resolve_budget(event)) + except Exception: + logger.exception( + "formatter failed for %s; falling back to legacy", event.category + ) + # ---- existing passthrough + Mode-B unchanged below ---- + if event.data and event.data.get("_meshai_precomposed") and event.title: return event.title diff --git a/work/meshai/notifications/shadow.py b/work/meshai/notifications/shadow.py new file mode 100644 index 0000000..4786a58 --- /dev/null +++ b/work/meshai/notifications/shadow.py @@ -0,0 +1,221 @@ +"""Phase-0b dry-run shadow comparator. + +Gated by env var MESHAI_SHADOW_CATEGORIES (comma-separated category names). +Empty/unset → fully OFF (zero overhead, zero side-effects on the real path). + +Safety contract — this module MUST uphold all of the following: + * NEVER calls GateResult.commit() + * NEVER calls bus.emit() + * NEVER writes to any DB table + * NEVER returns a value that the caller uses for a routing decision + * All exceptions swallowed + logged at DEBUG (shadow must never affect production) + * Filesystem writes only to _SHADOW_DIR (append-only JSONL, never read-back) + +Both public entry points — shadow_gate and shadow_render — return None always. +""" +from __future__ import annotations + +import functools +import json +import logging +import os +import time +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from meshai.notifications.events import Event + +logger = logging.getLogger("meshai.notifications.shadow") + +# JSONL output directory (inside the container data volume). +_SHADOW_DIR = "/app/data/shadow" + + +# --------------------------------------------------------------------------- +# Env-var gate +# --------------------------------------------------------------------------- + +@functools.lru_cache(maxsize=1) +def _load_shadow_categories() -> frozenset: + """Parse MESHAI_SHADOW_CATEGORIES into a frozenset. Cached after first call. + + Call _clear_enabled_cache() (e.g. from tests) to force re-parse after + mutating the env var. + """ + val = os.environ.get("MESHAI_SHADOW_CATEGORIES", "") + if not val.strip(): + return frozenset() + return frozenset(c.strip() for c in val.split(",") if c.strip()) + + +def _clear_enabled_cache() -> None: + """Clear the lru_cache so tests can mutate MESHAI_SHADOW_CATEGORIES.""" + _load_shadow_categories.cache_clear() + + +def enabled_for(category: str) -> bool: + """Return True iff shadow comparison is enabled for *category*. + + Fast path: returns False immediately when the env var is unset. + """ + return category in _load_shadow_categories() + + +# --------------------------------------------------------------------------- +# Internal JSONL writer — never raises, never blocks the caller +# --------------------------------------------------------------------------- + +def _append_jsonl(category: str, record: dict) -> None: + """Append one JSON line to /app/data/shadow/.jsonl. + + Any I/O failure is logged at DEBUG and silently discarded. + """ + try: + os.makedirs(_SHADOW_DIR, exist_ok=True) + path = os.path.join(_SHADOW_DIR, f"{category}.jsonl") + line = json.dumps(record, default=str) + "\n" + with open(path, "a", encoding="utf-8") as fh: + fh.write(line) + except Exception: + logger.debug( + "shadow: JSONL write failed for category=%s", category, exc_info=True + ) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def shadow_gate( + category: str, + data: dict, + *, + source: str, + now: float, + old_broadcast: bool = False, +) -> None: + """Dry-run the new gating decider for *category* and log mismatches. + + Called from consumer._normalize, BEFORE the default-deny return. + The old_broadcast flag is synthesized is not None (True = would broadcast). + The old suffixes (_severity_override, _dedup_suffix, _cooldown_suffix) are + read from *data* — they are the values the legacy handler stamped. + + Contract: + * NEVER calls GateResult.commit() + * NEVER emits any event + * NEVER writes any DB table + * All exceptions swallowed; returns None always + """ + if not enabled_for(category): + return + try: + from meshai.notifications.gating import get_decider + decider = get_decider(category) + if decider is None: + # No decider registered yet (Phase 0 — all DECIDERS are empty). + # Nothing to compare; exit silently. + return + + # ---- Phase 1+: a real decider is registered ---- + # Extract old-path suffixes from the data dict the handler populated. + old_sev = (data or {}).get("_severity_override") + old_dedup = (data or {}).get("_dedup_suffix") + old_cool = (data or {}).get("_cooldown_suffix") + + # Call the new decider. data serves as a proxy for the full Event here + # (both carry _dedup_suffix, _cooldown_suffix, _severity_override). + # Phase 1 deciders that require a full Event object will need Hook 1 + # to be relocated to post-normalize; Phase 0 never reaches here. + try: + new_result = decider(data, now=now) + except Exception: + logger.debug( + "shadow_gate: decider raised for category=%s", category, exc_info=True + ) + return + + # SAFETY CHECKPOINT: we inspect new_result but NEVER call .commit(). + new_broadcast = new_result.broadcast if new_result is not None else False + new_patch = (new_result.data_patch or {}) if new_result is not None else {} + + if old_broadcast != new_broadcast: + record = { + "tag": "SHADOW_MISMATCH", + "type": "gate", + "category": category, + "source": source, + "at": now, + "old_broadcast": old_broadcast, + "new_broadcast": new_broadcast, + "old_severity_override": old_sev, + "old_dedup_suffix": old_dedup, + "old_cooldown_suffix": old_cool, + "new_severity_override": new_patch.get("_severity_override"), + "new_dedup_suffix": new_patch.get("_dedup_suffix"), + "new_cooldown_suffix": new_patch.get("_cooldown_suffix"), + "lifecycle": getattr(new_result, "lifecycle", ""), + "reason": getattr(new_result, "reason", ""), + } + _append_jsonl(category, record) + logger.debug( + "shadow_gate MISMATCH cat=%s old=%s new=%s reason=%s", + category, old_broadcast, new_broadcast, + getattr(new_result, "reason", ""), + ) + except Exception: + # Belt-and-suspenders: any unexpected error is silently absorbed. + logger.debug( + "shadow_gate: unhandled error for category=%s", category, exc_info=True + ) + + +def shadow_render(category: str, event: "Event", *, old_wire: str) -> None: + """Dry-run the new formatter for *category* and log wire-string mismatches. + + Called from dispatcher._dispatch_toggle_path, after compose_mesh_message + has already produced *old_wire*. The real path keeps using old_wire + unchanged; this function only observes. + + Contract: + * NEVER modifies old_wire or event + * NEVER emits, commits, or writes any DB table + * All exceptions swallowed; returns None always + """ + if not enabled_for(category): + return + try: + from meshai.notifications.formatters import get_formatter + formatter = get_formatter(category) + if formatter is None: + # No formatter registered yet (Phase 0 — all FORMATTERS are empty). + return + + # ---- Phase 1+: a real formatter is registered ---- + from meshai.notifications.formatters._budget import budget_for + budget = budget_for(getattr(event, "source", "") or "") + try: + new_wire = formatter(event, now=time.time(), budget=budget) + except Exception: + logger.debug( + "shadow_render: formatter raised for category=%s", + category, exc_info=True, + ) + return + + if new_wire != old_wire: + record = { + "tag": "SHADOW_MISMATCH", + "type": "render", + "category": category, + "event_id": getattr(event, "id", None), + "at": time.time(), + "old_wire": old_wire, + "new_wire": new_wire, + } + _append_jsonl(category, record) + logger.debug("shadow_render MISMATCH cat=%s", category) + except Exception: + logger.debug( + "shadow_render: unhandled error for category=%s", category, exc_info=True + ) diff --git a/work/requirements.txt b/work/requirements.txt index 62acfbc..091bb36 100644 --- a/work/requirements.txt +++ b/work/requirements.txt @@ -15,3 +15,4 @@ fastapi>=0.110.0 uvicorn[standard]>=0.27.0 aiomqtt>=2.0.0 sgp4>=2.22 +tzdata diff --git a/work/scripts/capture_fixtures.py b/work/scripts/capture_fixtures.py new file mode 100644 index 0000000..5a766c0 --- /dev/null +++ b/work/scripts/capture_fixtures.py @@ -0,0 +1,244 @@ +"""Read-only ephemeral fixture capture from NATS JetStream. + +Captures real Central CloudEvents envelopes WITHOUT disturbing the live +durable consumers by using an ephemeral pull consumer (no durable name, +AckPolicy.none, short inactive_threshold for auto-deletion). + +Run from inside the meshai container:: + + docker exec meshai python /app/scripts/capture_fixtures.py \\ + --hazard earthquake_event \\ + --subject "central.usgs_quake.>" \\ + --mode all --max 20 + + # Dry-run (count only, no file writes): + docker exec meshai python /app/scripts/capture_fixtures.py \\ + --hazard earthquake_event \\ + --subject "central.usgs_quake.>" \\ + --mode all --max 20 --dry-run + + # Last-per-subject snapshot: + docker exec meshai python /app/scripts/capture_fixtures.py \\ + --hazard nws \\ + --subject "central.nws.>" \\ + --mode last + +Modes +----- +--mode last DeliverPolicy.LAST_PER_SUBJECT — one message per subject key. + Useful for a current-state snapshot. +--mode all DeliverPolicy.ALL — bounded history. REQUIRED: --max N cap + to avoid pulling 330k+ traffic messages. + +Output +------ +Each captured envelope is written as:: + + tests/fixtures//.json + { + "envelope": { ... }, # raw Central CloudEvents payload + "subject": "central.usgs_quake.us7000xyz", + "captured_epoch": 1750000000 + } + +Safety +------ +The ephemeral consumer is created with AckPolicy.none and a 30-second +inactive_threshold. It is never assigned a durable name, so it never +advances the live durable consumers' sequence pointers and is automatically +cleaned up by the NATS server after inactivity. +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import pathlib +import sys +import time + +# -------------------------------------------------------------------------- +# All network + config access is deferred to main() so this module is safely +# importable in unit-test environments without a running NATS server. +# -------------------------------------------------------------------------- + + +def _output_dir(hazard: str) -> pathlib.Path: + """Resolve tests/fixtures// relative to the repo root.""" + # Script lives at /scripts/capture_fixtures.py; + # fixtures live at /tests/fixtures//. + repo_root = pathlib.Path(__file__).parent.parent + return repo_root / "tests" / "fixtures" / hazard + + +async def _run( + *, + nats_url: str, + stream: str, + subject: str, + hazard: str, + mode: str, + max_msgs: int, + dry_run: bool, +) -> int: + """Connect, create ephemeral consumer, pull messages, write fixtures. + + Returns the count of messages captured (or counted, for --dry-run). + """ + import nats + from nats.js.api import AckPolicy, ConsumerConfig, DeliverPolicy + + nc = await nats.connect(nats_url) + try: + js = nc.jetstream() + + # Build an ephemeral consumer config (no durable_name = ephemeral). + # AckPolicy.none avoids needing to ack — purely read-only. + # inactive_threshold of 30 s ensures the NATS server auto-deletes it. + deliver_policy = ( + DeliverPolicy.LAST_PER_SUBJECT + if mode == "last" + else DeliverPolicy.ALL + ) + cfg = ConsumerConfig( + # durable_name intentionally omitted → ephemeral consumer + filter_subject=subject, + deliver_policy=deliver_policy, + ack_policy=AckPolicy.NONE, + inactive_threshold=30.0, # seconds → server auto-deletes after idle + ) + + # Create ephemeral pull consumer (server-side, no local binding name). + consumer_info = await js.add_consumer(stream, cfg) + consumer_name = consumer_info.name + + out_dir = _output_dir(hazard) + if not dry_run: + out_dir.mkdir(parents=True, exist_ok=True) + + captured = 0 + fetch_batch = min(max_msgs, 50) # pull in bounded batches + + while captured < max_msgs: + batch = min(fetch_batch, max_msgs - captured) + try: + msgs = await js.pull_subscribe_bind( + stream, consumer_name + ).fetch(batch, timeout=5.0) + except nats.errors.TimeoutError: + break # no more messages within timeout + + if not msgs: + break + + for msg in msgs: + try: + envelope = json.loads(msg.data) + except Exception: + continue # skip unparseable messages + + if dry_run: + captured += 1 + print( + f" [dry-run] #{captured} subject={msg.subject!r}", + file=sys.stderr, + ) + else: + record = { + "envelope": envelope, + "subject": msg.subject, + "captured_epoch": int(time.time()), + } + out_path = out_dir / f"{captured:04d}.json" + out_path.write_text( + json.dumps(record, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + captured += 1 + print( + f" wrote {out_path.relative_to(pathlib.Path.cwd())} " + f"subject={msg.subject!r}", + file=sys.stderr, + ) + + if captured >= max_msgs: + break + + # For last-per-subject: a single fetch is sufficient. + if mode == "last": + break + + # Delete the ephemeral consumer explicitly (belt-and-suspenders). + try: + await js.delete_consumer(stream, consumer_name) + except Exception: + pass # server already cleaned up, or error is non-fatal + + return captured + + finally: + await nc.drain() + await nc.close() + + +def _load_nats_url() -> str: + """Read the NATS URL from meshai config or env override.""" + # Allow an explicit env override for CI / ad-hoc use. + if "MESHAI_NATS_URL" in os.environ: + return os.environ["MESHAI_NATS_URL"] + try: + from meshai.config_loader import load_config + cfg = load_config() + return cfg.environmental.central.url # type: ignore[attr-defined] + except Exception: + return "nats://localhost:4222" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Capture Central NATS envelopes as fixture files (read-only)." + ) + parser.add_argument("--hazard", required=True, + help="Hazard category label (used as fixture sub-dir).") + parser.add_argument("--subject", required=True, + help="NATS subject filter, e.g. 'central.usgs_quake.>'.") + parser.add_argument("--stream", default="CENTRAL", + help="JetStream stream name (default: CENTRAL).") + parser.add_argument("--mode", choices=["last", "all"], default="all", + help="DeliverPolicy: last=LAST_PER_SUBJECT, all=ALL (default: all).") + parser.add_argument("--max", type=int, default=50, dest="max_msgs", + help="Maximum messages to capture (required cap; default: 50).") + parser.add_argument("--nats-url", default=None, + help="Override the NATS URL (default: read from meshai config).") + parser.add_argument("--dry-run", action="store_true", + help="Count messages only; do not write fixture files.") + args = parser.parse_args(argv) + + nats_url = args.nats_url or _load_nats_url() + print( + f"capture_fixtures: url={nats_url!r} stream={args.stream!r} " + f"subject={args.subject!r} hazard={args.hazard!r} " + f"mode={args.mode!r} max={args.max_msgs} dry_run={args.dry_run}", + file=sys.stderr, + ) + + count = asyncio.run( + _run( + nats_url=nats_url, + stream=args.stream, + subject=args.subject, + hazard=args.hazard, + mode=args.mode, + max_msgs=args.max_msgs, + dry_run=args.dry_run, + ) + ) + + verb = "counted" if args.dry_run else "captured" + print(f"{verb} {count} envelope(s) for hazard={args.hazard!r}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/work/tests/conftest.py b/work/tests/conftest.py index a7b3a85..c26c240 100644 --- a/work/tests/conftest.py +++ b/work/tests/conftest.py @@ -12,13 +12,47 @@ and clears the persistence-layer threading.local caches around each test. Existing tests that don't reference any fixture get isolation for free; tests that explicitly use a `db_path` (or similar) fixture can still override the env var inside their own fixture body -- last setenv wins. + +Phase-0 addition: a session-scoped TZ fixture validates that +America/Boise resolves via zoneinfo, exercising the tzdata guard needed +in CI environments. The fixture saves and restores the original TZ so +that existing tests running with naive datetimes are unaffected. """ +import os +import time +import zoneinfo + import pytest from meshai.persistence import close_thread_connection from meshai.persistence import db as _persistence_db +@pytest.fixture(scope="session", autouse=True) +def _tz_boise(): + """Set TZ=America/Boise, call tzset(), assert zoneinfo resolves, then restore. + + This validates that the tzdata package (or system zoneinfo) is available + in CI before any formatter/renderer test runs. The original TZ is restored + so that existing tests using naive datetimes are not affected by the + Mountain Time offset. + """ + original_tz = os.environ.get("TZ") + os.environ["TZ"] = "America/Boise" + time.tzset() + # Assert that zoneinfo can resolve the timezone (requires tzdata package + # or /usr/share/zoneinfo/America/Boise on the system). + zoneinfo.ZoneInfo("America/Boise") + # Restore the original TZ so that tests using naive datetime / time.localtime() + # behave the same as they did before this session fixture was added. + if original_tz is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = original_tz + time.tzset() + yield + + @pytest.fixture(autouse=True) def _isolate_meshai_db(tmp_path, monkeypatch): """Point MESHAI_DB_PATH at a tmp file per test + run init_db so the diff --git a/work/tests/harness/__init__.py b/work/tests/harness/__init__.py new file mode 100644 index 0000000..578878e --- /dev/null +++ b/work/tests/harness/__init__.py @@ -0,0 +1,32 @@ +"""Phase-0b test harness — golden + gate-sequence helpers. + +Import surface: + + from tests.harness import ( + pinned_time, + pinned_tz, + render_golden, + assert_byte_identical, + load_fixtures, + run_gate_sequence, + ) + +All helpers are pure-Python, stdlib-only, and safe to import at collection time. +""" +from tests.harness.goldens import ( + assert_byte_identical, + load_fixtures, + pinned_time, + pinned_tz, + render_golden, + run_gate_sequence, +) + +__all__ = [ + "assert_byte_identical", + "load_fixtures", + "pinned_time", + "pinned_tz", + "render_golden", + "run_gate_sequence", +] diff --git a/work/tests/harness/goldens.py b/work/tests/harness/goldens.py new file mode 100644 index 0000000..a0db64e --- /dev/null +++ b/work/tests/harness/goldens.py @@ -0,0 +1,284 @@ +"""Phase-0b golden + gate-sequence test helpers. + +pinned_time(epoch) + Context manager: monkeypatches meshai.notifications.clock.now / now_dt to + a fixed epoch so golden files are byte-stable. SCOPED — does not touch + the process TZ. Restores the originals on exit (exception-safe). + +pinned_tz(name) + Separate opt-in context manager: sets os.environ['TZ'] and calls + time.tzset(), restores on exit. Only golden tests that format %Z need + this; enabling it globally would break naive-datetime tests. + +render_golden(handler_render_fn, envelope, *, at) + Run handler_render_fn(envelope) under pinned_time(at) and return the wire + string. Used to CAPTURE the baseline golden (not to assert — compare + against assert_byte_identical later). + +assert_byte_identical(new, golden) + Compare two strings by their UTF-8 byte sequences (len + content). On + mismatch emit a readable unified diff and raise AssertionError. + +load_fixtures(hazard) + Read every tests/fixtures//*.json file (sorted) and return a list + of dicts, each with keys {envelope, subject, captured_epoch}. + +run_gate_sequence(old_handler, new_decider, ordered_fixtures, *, timeline) + Stub comparator: exercise both the old and new gating path across an + explicit now timeline and return a diff report. + + old_handler : callable(fixture: dict, *, now: float) -> bool + True = old path would broadcast, False = suppress. + new_decider : callable(fixture: dict, *, now: float) -> GateResult | None + May return None (no decision) → treated as suppress. + ordered_fixtures : list[dict] — fixtures in replay order + timeline : list[float] — one epoch per fixture (len must match) + + Returns list[dict] with keys: + fixture_n : int — 0-based index into ordered_fixtures + now : float + old_broadcast : bool + new_broadcast : bool + match : bool — True when both paths agree + diffs : dict — empty when match; keys describe divergence +""" +from __future__ import annotations + +import difflib +import json +import os +import pathlib +import time +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Callable, Optional + + +# --------------------------------------------------------------------------- +# pinned_time — deterministic clock seam for golden tests +# --------------------------------------------------------------------------- + +@contextmanager +def pinned_time(epoch: float): + """Monkeypatch meshai.notifications.clock.now / now_dt to *epoch*. + + Restores the originals on exit even if the body raises. + Does NOT touch TZ — use pinned_tz for %Z-sensitive golden tests. + + Usage:: + + with pinned_time(1_700_000_000.0): + result = some_handler(envelope) + # result is deterministic for golden comparison + """ + import meshai.notifications.clock as _clock_mod + _orig_now = _clock_mod.now + _orig_now_dt = _clock_mod.now_dt + _clock_mod.now = lambda: float(epoch) + _clock_mod.now_dt = lambda tz=None: datetime.fromtimestamp( + epoch, tz=(tz if tz is not None else timezone.utc) + ) + try: + yield epoch + finally: + _clock_mod.now = _orig_now + _clock_mod.now_dt = _orig_now_dt + + +# --------------------------------------------------------------------------- +# pinned_tz — opt-in TZ override (only for %Z-sensitive golden tests) +# --------------------------------------------------------------------------- + +@contextmanager +def pinned_tz(name: str = "America/Boise"): + """Set TZ=*name*, call time.tzset(), restore on exit. + + Opt-in only: enabling TZ globally breaks existing tests that format + naive datetimes or rely on UTC. Golden tests that include a %Z token + in their expected output should wrap only that call in pinned_tz. + + Usage:: + + with pinned_tz("America/Boise"): + with pinned_time(1_700_000_000.0): + result = handler(envelope) + """ + orig = os.environ.get("TZ") + os.environ["TZ"] = name + time.tzset() + try: + yield name + finally: + if orig is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = orig + time.tzset() + + +# --------------------------------------------------------------------------- +# render_golden — capture a wire string under frozen time +# --------------------------------------------------------------------------- + +def render_golden( + handler_render_fn: Callable, + envelope: dict, + *, + at: float, +) -> str: + """Return the wire string from handler_render_fn(envelope) with time frozen. + + handler_render_fn : callable(envelope: dict) -> str + Typically a lambda wrapper around a handler function, e.g.:: + + render_golden( + lambda env: handle_nws(env, subject, data={}), + envelope, + at=1_700_000_000.0, + ) + + The result is used as the *golden* baseline. Store it (e.g. in a .txt + fixture) and compare future renders with assert_byte_identical. + """ + with pinned_time(at): + return handler_render_fn(envelope) + + +# --------------------------------------------------------------------------- +# assert_byte_identical — byte-level golden comparison +# --------------------------------------------------------------------------- + +def assert_byte_identical(new: str, golden: str) -> None: + """Assert that *new* and *golden* are identical as UTF-8 byte sequences. + + Checks both len(x.encode('utf-8')) and content. On mismatch emits a + unified diff and raises AssertionError with a readable message. + """ + new_bytes = new.encode("utf-8") + golden_bytes = golden.encode("utf-8") + if new_bytes == golden_bytes: + return + # Build a readable unified diff (character-level, single-line strings). + diff_lines = list( + difflib.unified_diff( + [golden], + [new], + fromfile="golden", + tofile="new", + lineterm="", + ) + ) + diff_str = "\n".join(diff_lines) if diff_lines else "" + raise AssertionError( + f"assert_byte_identical failed:\n" + f" golden: {len(golden_bytes)} bytes\n" + f" new: {len(new_bytes)} bytes\n" + f"{diff_str}" + ) + + +# --------------------------------------------------------------------------- +# load_fixtures — read fixture JSON files for a hazard category +# --------------------------------------------------------------------------- + +def load_fixtures(hazard: str) -> list[dict]: + """Read all tests/fixtures//*.json files (sorted) and return as list. + + Each JSON file is expected to contain:: + + { + "envelope": {...}, # raw Central CloudEvents envelope + "subject": "...", # NATS subject + "captured_epoch": 1234567 # wall-clock at capture time + } + + Returns an empty list if the directory does not exist (no fixtures yet). + """ + fixtures_dir = ( + pathlib.Path(__file__).parent.parent / "fixtures" / hazard + ) + if not fixtures_dir.is_dir(): + return [] + result = [] + for p in sorted(fixtures_dir.glob("*.json")): + with open(p, encoding="utf-8") as fh: + result.append(json.load(fh)) + return result + + +# --------------------------------------------------------------------------- +# run_gate_sequence — old-vs-new broadcast decision comparator +# --------------------------------------------------------------------------- + +def run_gate_sequence( + old_handler: Callable, + new_decider: Callable, + ordered_fixtures: list, + *, + timeline: list, +) -> list[dict]: + """Compare broadcast/suppress decisions of old_handler vs new_decider. + + Parameters + ---------- + old_handler: + callable(fixture: dict, *, now: float) -> bool + True = old path would broadcast + False = old path would suppress + new_decider: + callable(fixture: dict, *, now: float) -> GateResult | None + None → treated as suppress (False). + The returned GateResult.commit is NEVER called here. + ordered_fixtures: + list of fixture dicts in replay order. + timeline: + list[float] — one epoch per fixture. Must be same length as + ordered_fixtures; ValueError raised otherwise. + + Returns + ------- + list[dict] — one entry per fixture: + fixture_n : int — 0-based index + now : float — epoch used for this step + old_broadcast : bool + new_broadcast : bool + match : bool — True when old_broadcast == new_broadcast + diffs : dict — empty when match; populated fields on divergence: + "broadcast": {"old": bool, "new": bool} + """ + if len(ordered_fixtures) != len(timeline): + raise ValueError( + f"run_gate_sequence: ordered_fixtures has {len(ordered_fixtures)} entries " + f"but timeline has {len(timeline)} — they must match." + ) + + results = [] + for i, (fixture, now_ts) in enumerate(zip(ordered_fixtures, timeline)): + # --- old path --- + old_broadcast = bool(old_handler(fixture, now=now_ts)) + + # --- new path (never commit) --- + new_result = None + try: + new_result = new_decider(fixture, now=now_ts) + except Exception: + pass # treat a crashing decider as suppress + new_broadcast = (new_result.broadcast if new_result is not None else False) + + # --- diff --- + diffs: dict = {} + if old_broadcast != new_broadcast: + diffs["broadcast"] = {"old": old_broadcast, "new": new_broadcast} + + results.append( + { + "fixture_n": i, + "now": now_ts, + "old_broadcast": old_broadcast, + "new_broadcast": new_broadcast, + "match": not diffs, + "diffs": diffs, + } + ) + + return results diff --git a/work/tests/test_budget_shim.py b/work/tests/test_budget_shim.py new file mode 100644 index 0000000..540a886 --- /dev/null +++ b/work/tests/test_budget_shim.py @@ -0,0 +1,35 @@ +"""Phase-0: verify the budget.py re-export shim is identity-equal to the impl. + +Both import paths must resolve to the SAME function objects so any runtime +patching (e.g. in existing tests) affects both paths simultaneously. +""" + +import meshai.central.budget as _shim +import meshai.notifications.formatters._budget as _impl + + +def test_budget_for_is_same_object(): + assert _shim.budget_for is _impl.budget_for, ( + "meshai.central.budget.budget_for must be the same object as " + "meshai.notifications.formatters._budget.budget_for" + ) + + +def test_fit_to_budget_is_same_object(): + assert _shim.fit_to_budget is _impl.fit_to_budget, ( + "meshai.central.budget.fit_to_budget must be the same object as " + "meshai.notifications.formatters._budget.fit_to_budget" + ) + + +def test_shim_imports_work(): + """Smoke: the public names are importable from the legacy path.""" + from meshai.central.budget import budget_for, fit_to_budget # noqa: F401 + assert callable(budget_for) + assert callable(fit_to_budget) + + +def test_budget_for_returns_default_without_adapter_config(monkeypatch): + """budget_for falls back to 140 when the adapter key is absent.""" + val = _shim.budget_for("__no_such_adapter__") + assert val == 140 diff --git a/work/tests/test_capture_script_importable.py b/work/tests/test_capture_script_importable.py new file mode 100644 index 0000000..dec48f2 --- /dev/null +++ b/work/tests/test_capture_script_importable.py @@ -0,0 +1,64 @@ +"""Phase-0b: verify capture_fixtures.py is importable without a NATS connection. + +This test guards against syntax errors, bad top-level imports, or any +network access at import time. It does NOT execute main() or connect to NATS. +""" +from __future__ import annotations + +import importlib.util +import pathlib +import sys +import types + + +_SCRIPT_PATH = ( + pathlib.Path(__file__).parent.parent / "scripts" / "capture_fixtures.py" +) + + +class TestCaptureScriptImportable: + def test_script_file_exists(self): + assert _SCRIPT_PATH.is_file(), ( + f"Expected capture_fixtures.py at {_SCRIPT_PATH} but file not found" + ) + + def test_script_importable_without_network(self): + """Load the script as a module; must succeed without NATS connection.""" + spec = importlib.util.spec_from_file_location( + "capture_fixtures_script", _SCRIPT_PATH + ) + module = importlib.util.module_from_spec(spec) + # Execute the module body — must not connect to anything. + spec.loader.exec_module(module) # type: ignore[union-attr] + + def test_main_function_is_callable(self): + """The script must expose a main() callable.""" + spec = importlib.util.spec_from_file_location( + "capture_fixtures_script_main", _SCRIPT_PATH + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # type: ignore[union-attr] + assert callable(getattr(module, "main", None)), ( + "capture_fixtures.py must expose a main() callable" + ) + + def test_output_dir_helper_is_callable(self): + """_output_dir is a helper for resolving fixture paths; must be importable.""" + spec = importlib.util.spec_from_file_location( + "capture_fixtures_script_dir", _SCRIPT_PATH + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # type: ignore[union-attr] + assert callable(getattr(module, "_output_dir", None)) + + def test_help_flag_is_parseable(self): + """main(['--help']) must raise SystemExit(0), not a real error.""" + spec = importlib.util.spec_from_file_location( + "capture_fixtures_script_help", _SCRIPT_PATH + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # type: ignore[union-attr] + import pytest + with pytest.raises(SystemExit) as exc_info: + module.main(["--help"]) + assert exc_info.value.code == 0 diff --git a/work/tests/test_clock_seam.py b/work/tests/test_clock_seam.py new file mode 100644 index 0000000..4922ef2 --- /dev/null +++ b/work/tests/test_clock_seam.py @@ -0,0 +1,59 @@ +"""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)}" + ) diff --git a/work/tests/test_formatter_scaffold.py b/work/tests/test_formatter_scaffold.py new file mode 100644 index 0000000..04644b2 --- /dev/null +++ b/work/tests/test_formatter_scaffold.py @@ -0,0 +1,66 @@ +"""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", [ + "weather_warning", + "earthquake_event", + "wildfire_incident", + "road_closure", + "battery_critical", +]) +def test_get_formatter_returns_none_while_registry_empty(category): + """FORMATTERS is empty; get_formatter must return None for any real category.""" + # Guarantee the registry is empty for these categories (it starts empty at + # module level; tests run in isolation from each other's registrations). + assert category not in FORMATTERS, ( + f"Category {category!r} should not be in FORMATTERS at Phase 0" + ) + 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. + """ + # 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) diff --git a/work/tests/test_formatters_no_clock.py b/work/tests/test_formatters_no_clock.py new file mode 100644 index 0000000..3d3d4bd --- /dev/null +++ b/work/tests/test_formatters_no_clock.py @@ -0,0 +1,46 @@ +"""Phase-0 purity guard: formatter modules must not call time directly. + +Every file under meshai/notifications/formatters/ must route all time reads +through meshai.notifications.clock (the determinism seam). Direct calls to +datetime.now or time.time( inside a formatter break the monkeypatch contract. + +This test is intentionally static (source-text check) so it catches new +formatters that forget the contract even before any test calls them. +""" + +import glob +import os + + +def _formatter_sources(): + """Return (path, source) pairs for all .py files in the formatters package.""" + pattern = os.path.join( + os.path.dirname(__file__), + "..", "meshai", "notifications", "formatters", "*.py", + ) + paths = sorted(glob.glob(pattern)) + return [(p, open(p).read()) for p in paths] + + +def test_no_datetime_now_in_formatters(): + """No formatter file may call datetime.now directly.""" + violations = [] + for path, src in _formatter_sources(): + if "datetime.now" in src: + violations.append(os.path.basename(path)) + assert not violations, ( + "The following formatter files call datetime.now directly " + "(use clock.now_dt() instead):\n" + "\n".join(violations) + ) + + +def test_no_time_time_in_formatters(): + """No formatter file may call time.time( directly.""" + violations = [] + for path, src in _formatter_sources(): + if "time.time(" in src: + violations.append(os.path.basename(path)) + assert not violations, ( + "The following formatter files call time.time( directly " + "(use clock.now() instead):\n" + "\n".join(violations) + ) diff --git a/work/tests/test_harness_selftest.py b/work/tests/test_harness_selftest.py new file mode 100644 index 0000000..fe67ac3 --- /dev/null +++ b/work/tests/test_harness_selftest.py @@ -0,0 +1,298 @@ +"""Phase-0b: self-test the harness helpers. + +Verifies: + 1. pinned_time freezes clock.now() to the given epoch. + 2. pinned_time restores the original after the block. + 3. pinned_tz sets and restores TZ (opt-in; tested in isolation). + 4. assert_byte_identical passes on equal strings, fails on differing ones. + 5. run_gate_sequence diffs correctly with two simple fake deciders. + 6. load_fixtures returns empty list for a non-existent hazard. +""" +from __future__ import annotations + +import os +import time + +import pytest + +import meshai.notifications.clock as clock_mod +from tests.harness import ( + assert_byte_identical, + load_fixtures, + pinned_time, + pinned_tz, + run_gate_sequence, +) + + +# --------------------------------------------------------------------------- +# GateResult import for fake deciders +# --------------------------------------------------------------------------- + +from meshai.notifications.gating.base import GateResult + + +# --------------------------------------------------------------------------- +# 1-2. pinned_time +# --------------------------------------------------------------------------- + +_FROZEN = 1_700_000_000.0 + + +class TestPinnedTime: + def test_clock_now_is_frozen_inside_block(self): + with pinned_time(_FROZEN): + result = clock_mod.now() + assert result == _FROZEN + + def test_clock_now_dt_is_frozen_inside_block(self): + from datetime import timezone + with pinned_time(_FROZEN): + dt = clock_mod.now_dt(tz=timezone.utc) + assert dt.timestamp() == pytest.approx(_FROZEN) + + def test_clock_now_restored_after_block(self): + before = clock_mod.now + with pinned_time(_FROZEN): + pass + # The function object is restored to the original callable. + assert clock_mod.now is before + + def test_clock_now_restored_on_exception(self): + original_now = clock_mod.now + try: + with pinned_time(_FROZEN): + raise RuntimeError("test exception") + except RuntimeError: + pass + # Must be restored even after exception. + assert clock_mod.now is original_now + + def test_nested_pinned_time_restores_correctly(self): + outer_epoch = 1_600_000_000.0 + inner_epoch = 1_700_000_000.0 + with pinned_time(outer_epoch): + assert clock_mod.now() == outer_epoch + with pinned_time(inner_epoch): + assert clock_mod.now() == inner_epoch + # outer is restored + assert clock_mod.now() == outer_epoch + + def test_pinned_time_yields_epoch(self): + with pinned_time(_FROZEN) as yielded: + assert yielded == _FROZEN + + +# --------------------------------------------------------------------------- +# 3. pinned_tz +# --------------------------------------------------------------------------- + +class TestPinnedTz: + def test_sets_tz_inside_block(self): + with pinned_tz("UTC"): + assert os.environ.get("TZ") == "UTC" + + def test_restores_tz_after_block(self): + orig = os.environ.get("TZ") + with pinned_tz("UTC"): + pass + restored = os.environ.get("TZ") + assert restored == orig # both None, or both the original string + + def test_restores_tz_on_exception(self): + orig = os.environ.get("TZ") + try: + with pinned_tz("UTC"): + raise RuntimeError("test") + except RuntimeError: + pass + assert os.environ.get("TZ") == orig + + def test_yields_name(self): + with pinned_tz("America/Boise") as name: + assert name == "America/Boise" + + +# --------------------------------------------------------------------------- +# 4. assert_byte_identical +# --------------------------------------------------------------------------- + +class TestAssertByteIdentical: + def test_passes_on_equal_strings(self): + assert_byte_identical("hello world", "hello world") + + def test_passes_on_emoji_equal(self): + s = "🔥 FIRE: Test Creek NE 1500ac 35% routine" + assert_byte_identical(s, s) + + def test_fails_on_differing_content(self): + with pytest.raises(AssertionError) as exc_info: + assert_byte_identical("hello world!", "hello world") + msg = str(exc_info.value) + assert "golden" in msg or "new" in msg # diff headers present + + def test_fails_on_byte_length_difference(self): + # Same visible chars but different byte lengths via Unicode. + a = "café" # café (NFC, 2-byte é) + b = "café" # cafe + combining accent (decomposed, also visually café) + # These have different UTF-8 byte lengths. + if a.encode("utf-8") != b.encode("utf-8"): + with pytest.raises(AssertionError): + assert_byte_identical(a, b) + else: + # If the runtime normalized them, both forms happen to be equal. + assert_byte_identical(a, b) + + def test_diff_message_is_readable(self): + with pytest.raises(AssertionError) as exc_info: + assert_byte_identical("new string", "golden string") + msg = str(exc_info.value) + # Should mention byte counts. + assert "bytes" in msg + + def test_passes_on_empty_strings(self): + assert_byte_identical("", "") + + +# --------------------------------------------------------------------------- +# 5. run_gate_sequence with fake deciders +# --------------------------------------------------------------------------- + +class TestRunGateSequence: + """Use two simple fake deciders to verify the diff logic.""" + + # Fake old handler: broadcasts if the fixture has "broadcast": true. + @staticmethod + def _old_handler(fixture: dict, *, now: float) -> bool: + return bool(fixture.get("broadcast", True)) + + # Fake new decider: broadcasts if "new_broadcast" key is present and true. + @staticmethod + def _new_decider_agree(fixture: dict, *, now: float) -> GateResult: + # Mirrors the old handler. + broadcast = bool(fixture.get("broadcast", True)) + return GateResult(broadcast=broadcast, lifecycle="test", reason="agree") + + @staticmethod + def _new_decider_disagree(fixture: dict, *, now: float) -> GateResult: + # Always suppresses — will disagree when old handler broadcasts. + return GateResult(broadcast=False, lifecycle="test", reason="disagree") + + def _make_fixtures(self, specs): + """Build minimal fixture dicts.""" + return [{"broadcast": b} for b in specs] + + def _make_timeline(self, n): + return [1_700_000_000.0 + i * 300 for i in range(n)] + + def test_all_agree(self): + specs = [True, False, True] + fixtures = self._make_fixtures(specs) + timeline = self._make_timeline(len(fixtures)) + results = run_gate_sequence( + self._old_handler, + self._new_decider_agree, + fixtures, + timeline=timeline, + ) + assert len(results) == 3 + for r in results: + assert r["match"] is True, f"Expected match at fixture {r['fixture_n']}: {r}" + assert r["diffs"] == {} + + def test_disagree_is_flagged(self): + fixtures = self._make_fixtures([True, True, False]) + timeline = self._make_timeline(len(fixtures)) + results = run_gate_sequence( + self._old_handler, + self._new_decider_disagree, + fixtures, + timeline=timeline, + ) + # fixture 0 and 1: old=True, new=False → mismatch + assert results[0]["match"] is False + assert results[0]["diffs"]["broadcast"] == {"old": True, "new": False} + # fixture 2: old=False, new=False → match + assert results[2]["match"] is True + + def test_result_keys_present(self): + fixtures = self._make_fixtures([True]) + results = run_gate_sequence( + self._old_handler, + self._new_decider_agree, + fixtures, + timeline=self._make_timeline(1), + ) + r = results[0] + assert "fixture_n" in r + assert "now" in r + assert "old_broadcast" in r + assert "new_broadcast" in r + assert "match" in r + assert "diffs" in r + + def test_length_mismatch_raises(self): + with pytest.raises(ValueError, match="timeline"): + run_gate_sequence( + self._old_handler, + self._new_decider_agree, + [{"broadcast": True}, {"broadcast": False}], + timeline=[1_700_000_000.0], # length mismatch + ) + + def test_decider_exception_treated_as_suppress(self): + def _crashing_decider(fixture, *, now): + raise RuntimeError("boom") + + fixtures = self._make_fixtures([True]) + results = run_gate_sequence( + self._old_handler, + _crashing_decider, + fixtures, + timeline=self._make_timeline(1), + ) + # old=True, new=False (crashing decider → suppress) → mismatch + assert results[0]["old_broadcast"] is True + assert results[0]["new_broadcast"] is False + assert results[0]["match"] is False + + def test_now_is_passed_to_handlers(self): + received = [] + + def _handler_capturing_now(fixture, *, now): + received.append(("old", now)) + return True + + def _decider_capturing_now(fixture, *, now): + received.append(("new", now)) + return GateResult(broadcast=True) + + timeline = [1_700_000_000.0, 1_700_000_300.0] + run_gate_sequence( + _handler_capturing_now, + _decider_capturing_now, + [{"x": 1}, {"x": 2}], + timeline=timeline, + ) + assert ("old", 1_700_000_000.0) in received + assert ("old", 1_700_000_300.0) in received + assert ("new", 1_700_000_000.0) in received + + def test_empty_fixtures(self): + results = run_gate_sequence( + self._old_handler, + self._new_decider_agree, + [], + timeline=[], + ) + assert results == [] + + +# --------------------------------------------------------------------------- +# 6. load_fixtures for non-existent hazard +# --------------------------------------------------------------------------- + +class TestLoadFixtures: + def test_missing_hazard_returns_empty(self): + result = load_fixtures("__nonexistent_hazard_xyzzy__") + assert result == [] diff --git a/work/tests/test_import_smoke.py b/work/tests/test_import_smoke.py new file mode 100644 index 0000000..5fa0b1b --- /dev/null +++ b/work/tests/test_import_smoke.py @@ -0,0 +1,38 @@ +"""Phase-0 import smoke test: every meshai/central/*_handler.py module +must be importable without error after the budget shim refactor. + +This guards against a broken import chain (e.g. circular imports or a bad +re-export in the shim) that would silently break all handlers. +""" + +import glob +import importlib +import os + + +def _handler_modules(): + """Collect meshai.central.*_handler module names by globbing the source tree.""" + pattern = os.path.join( + os.path.dirname(__file__), + "..", "meshai", "central", "*_handler.py", + ) + paths = sorted(glob.glob(pattern)) + assert paths, "No *_handler.py files found — check the glob path" + modules = [] + for p in paths: + name = os.path.basename(p)[:-3] # strip .py + modules.append(f"meshai.central.{name}") + return modules + + +def test_all_central_handlers_importable(): + """Each *_handler module must import cleanly (no ImportError / circular deps).""" + failed = [] + for mod_name in _handler_modules(): + try: + importlib.import_module(mod_name) + except ImportError as exc: + failed.append(f"{mod_name}: {exc}") + assert not failed, ( + "The following handler modules raised ImportError:\n" + "\n".join(failed) + ) diff --git a/work/tests/test_shadow_inert.py b/work/tests/test_shadow_inert.py new file mode 100644 index 0000000..5177269 --- /dev/null +++ b/work/tests/test_shadow_inert.py @@ -0,0 +1,199 @@ +"""Phase-0b: verify shadow hooks are inert by default and never mutate state. + +Test cases: + 1. MESHAI_SHADOW_CATEGORIES unset → shadow_gate / shadow_render are pure no-ops + (no filesystem, no DB, no exception). + 2. MESHAI_SHADOW_CATEGORIES set to a category with no decider registered → + still no-op (get_decider returns None, early return). + 3. MESHAI_SHADOW_CATEGORIES set to a category with no formatter registered → + still no-op (get_formatter returns None, early return). + 4. shadow_gate / shadow_render never raise regardless of input. +""" +from __future__ import annotations + +import os +import pytest + +import meshai.notifications.shadow as shadow_mod + + +def _reset_shadow_cache(): + """Clear lru_cache so env-var changes take effect.""" + shadow_mod._clear_enabled_cache() + + +# --------------------------------------------------------------------------- +# Helper: a minimal fake Event-like object for shadow_render +# --------------------------------------------------------------------------- + +class _FakeEvent: + id = "test-event-001" + category = "earthquake_event" + source = "usgs_quake" + data: dict = {} + + +# --------------------------------------------------------------------------- +# Case 1: env var unset — must be completely off +# --------------------------------------------------------------------------- + +class TestShadowInertWhenEnvUnset: + """With MESHAI_SHADOW_CATEGORIES unset, all shadow functions are no-ops.""" + + def setup_method(self): + os.environ.pop("MESHAI_SHADOW_CATEGORIES", None) + _reset_shadow_cache() + + def test_enabled_for_returns_false(self): + assert shadow_mod.enabled_for("earthquake_event") is False + assert shadow_mod.enabled_for("nws") is False + assert shadow_mod.enabled_for("") is False + + def test_shadow_gate_no_filesystem(self, tmp_path, monkeypatch): + """shadow_gate must not touch the filesystem when the env var is unset.""" + monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow")) + shadow_mod.shadow_gate( + "earthquake_event", + {"_severity_override": "immediate"}, + source="usgs_quake", + now=1_700_000_000.0, + old_broadcast=True, + ) + # No shadow dir should have been created. + assert not (tmp_path / "shadow").exists() + + def test_shadow_render_no_filesystem(self, tmp_path, monkeypatch): + """shadow_render must not touch the filesystem when the env var is unset.""" + monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow")) + shadow_mod.shadow_render( + "earthquake_event", + _FakeEvent(), + old_wire="🌍 EQ M4.2: Test, ID 30mi NE routine", + ) + assert not (tmp_path / "shadow").exists() + + def test_shadow_gate_returns_none(self): + result = shadow_mod.shadow_gate( + "earthquake_event", {}, source="usgs", now=0.0, old_broadcast=False + ) + assert result is None + + def test_shadow_render_returns_none(self): + result = shadow_mod.shadow_render( + "earthquake_event", _FakeEvent(), old_wire="something" + ) + assert result is None + + +# --------------------------------------------------------------------------- +# Case 2: env var set but no decider registered for that category +# --------------------------------------------------------------------------- + +class TestShadowInertWhenNoDecider: + """MESHAI_SHADOW_CATEGORIES set but DECIDERS empty → still no-op.""" + + def setup_method(self): + os.environ["MESHAI_SHADOW_CATEGORIES"] = "earthquake_event" + _reset_shadow_cache() + + def teardown_method(self): + os.environ.pop("MESHAI_SHADOW_CATEGORIES", None) + _reset_shadow_cache() + + def test_enabled_for_returns_true(self): + assert shadow_mod.enabled_for("earthquake_event") is True + + def test_shadow_gate_no_filesystem_when_no_decider(self, tmp_path, monkeypatch): + """get_decider returns None → shadow_gate exits before any file write.""" + monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow")) + # DECIDERS is empty (Phase 0 scaffold), so get_decider("earthquake_event") + # returns None and shadow_gate returns early without writing anything. + shadow_mod.shadow_gate( + "earthquake_event", + {"_dedup_suffix": "M4.2"}, + source="usgs_quake", + now=1_700_000_000.0, + old_broadcast=True, + ) + assert not (tmp_path / "shadow").exists() + + def test_shadow_gate_does_not_raise(self): + """shadow_gate must not propagate any exception.""" + try: + shadow_mod.shadow_gate( + "earthquake_event", + None, # intentionally bad input — must not raise + source="usgs_quake", + now=1_700_000_000.0, + old_broadcast=False, + ) + except Exception as exc: + pytest.fail(f"shadow_gate raised unexpectedly: {exc!r}") + + +# --------------------------------------------------------------------------- +# Case 3: env var set but no formatter registered for that category +# --------------------------------------------------------------------------- + +class TestShadowRenderInertWhenNoFormatter: + """MESHAI_SHADOW_CATEGORIES set but FORMATTERS empty → still no-op.""" + + def setup_method(self): + os.environ["MESHAI_SHADOW_CATEGORIES"] = "earthquake_event" + _reset_shadow_cache() + + def teardown_method(self): + os.environ.pop("MESHAI_SHADOW_CATEGORIES", None) + _reset_shadow_cache() + + def test_shadow_render_no_filesystem_when_no_formatter(self, tmp_path, monkeypatch): + """get_formatter returns None → shadow_render exits before any file write.""" + monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow")) + shadow_mod.shadow_render( + "earthquake_event", + _FakeEvent(), + old_wire="old wire string", + ) + assert not (tmp_path / "shadow").exists() + + def test_shadow_render_does_not_raise(self): + """shadow_render must not propagate any exception.""" + try: + shadow_mod.shadow_render( + "earthquake_event", + None, # intentionally bad input — must not raise + old_wire="anything", + ) + except Exception as exc: + pytest.fail(f"shadow_render raised unexpectedly: {exc!r}") + + +# --------------------------------------------------------------------------- +# Case 4: multiple categories, partial enable +# --------------------------------------------------------------------------- + +class TestShadowPartialEnable: + """Only listed categories are enabled; others stay off.""" + + def setup_method(self): + os.environ["MESHAI_SHADOW_CATEGORIES"] = "nws,earthquake_event" + _reset_shadow_cache() + + def teardown_method(self): + os.environ.pop("MESHAI_SHADOW_CATEGORIES", None) + _reset_shadow_cache() + + def test_listed_category_is_enabled(self): + assert shadow_mod.enabled_for("nws") is True + assert shadow_mod.enabled_for("earthquake_event") is True + + def test_unlisted_category_is_disabled(self): + assert shadow_mod.enabled_for("fire") is False + assert shadow_mod.enabled_for("geomagnetic_storm") is False + + def test_shadow_gate_off_for_unlisted(self, tmp_path, monkeypatch): + monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow")) + shadow_mod.shadow_gate( + "fire", {}, source="wfigs", now=0.0, old_broadcast=True + ) + assert not (tmp_path / "shadow").exists()