"""Contract tests for scripts/replay_shoc_webhooks.py (plan Phase 5). Pins the replay tool's operator-safety and envelope contracts: dry-run is the default and performs NO HTTP, the sts account gate rejects any profile that does not resolve to seahaven-prod, the envelope data field lists stay byte-identical to the emitter's (the receiver sees one schema regardless of path), every replay envelope carries "replay": true, and delivery_id is deterministic across runs (receivers dedupe overlapping replays on it). The script is loaded by file path (scripts/ is not a package -- mirrors tests/test_reprocess_contract.py); boto3.Session is monkeypatched to a plain fake so main() runs fully offline. """ import importlib.util import json import pathlib import sys import pytest from tests.support import load_lambda_module # replay builds its boto3 session inside main(), so import is side-effect-free. _p = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "replay_shoc_webhooks.py" _spec = importlib.util.spec_from_file_location("replay_shoc_webhooks", _p) replay = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(replay) WO_ITEM = { "work_order_id": "11144580730", "wo_status": "assigned", "record_type": "new_work_order", "updated_at": "2026-07-01T00:00:00+00:00", "source_email_s3_key": "inbound/2026/07/abc.eml", } COMMENT_ITEM = { "work_order_id": "11144580730", "comment_id": "11144580730#2026-04-27T23:51:48#a1b2c3d4e5f6", "record_type": "comment", "commenter": "APM Technician", "text": "Vendor dispatched.", "created_at": "2026-04-27T23:51:48", "ingested_at": "2026-07-02T00:00:00+00:00", } # --- Offline fakes for main() ------------------------------------------------ class _FakeSTS: def __init__(self, account): self.account = account def get_caller_identity(self): return {"Account": self.account} class _FakeSecrets: def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name) return { "SecretString": json.dumps( {"keys": [{"kid": "2026-07-20T00", "secret": "a" * 64}]} ) } class _FakeWOTable: def get_item(self, Key): # noqa: N803 (boto3 kwarg name) return {"Item": dict(WO_ITEM, work_order_id=Key["work_order_id"])} class _FakeCommentsTable: def query(self, **kwargs): return {"Items": [dict(COMMENT_ITEM)]} class _FakeDynamo: def Table(self, name): # noqa: N802 (boto3 method name) if name == replay.WO_TABLE: return _FakeWOTable() return _FakeCommentsTable() class _FakeSession: def __init__(self, account): self.account = account def client(self, name): if name == "sts": return _FakeSTS(self.account) return _FakeSecrets() def resource(self, name): return _FakeDynamo() def _run_main(monkeypatch, account, argv): monkeypatch.setattr(replay.boto3, "Session", lambda **kwargs: _FakeSession(account)) def _no_http(*args, **kwargs): raise AssertionError("urlopen must not be called") monkeypatch.setattr(replay.urllib.request, "urlopen", _no_http) monkeypatch.setattr(sys, "argv", ["replay_shoc_webhooks.py", *argv]) replay.main() # --- Dry-run default + account gate ------------------------------------------ def test_dry_run_is_default_and_performs_no_http(monkeypatch, capsys): # No --execute: main() must complete without ever touching urlopen (the # patched opener raises if reached). _run_main( monkeypatch, replay.EXPECTED_ACCOUNT, ["--url", "https://receiver.invalid/webhook", "--work-order-id", "wo-1"], ) out = capsys.readouterr().out assert "DRY-RUN: would POST" in out assert "Dry run complete" in out def test_account_gate_rejects_wrong_account(monkeypatch): assert replay.EXPECTED_ACCOUNT == "011934824531" with pytest.raises(SystemExit) as exc: _run_main( monkeypatch, "999999999999", [ "--url", "https://receiver.invalid/webhook", "--work-order-id", "wo-1", ], ) assert "999999999999" in str(exc.value) assert replay.EXPECTED_ACCOUNT in str(exc.value) def test_selector_is_exactly_one_of_id_or_since(monkeypatch): with pytest.raises(SystemExit, match="exactly one"): _run_main( monkeypatch, replay.EXPECTED_ACCOUNT, ["--url", "https://receiver.invalid/webhook"], ) with pytest.raises(SystemExit, match="exactly one"): _run_main( monkeypatch, replay.EXPECTED_ACCOUNT, [ "--url", "https://receiver.invalid/webhook", "--work-order-id", "wo-1", "--since", "2026-07-24T02:00:00Z", ], ) # --- Envelope contract vs the emitter ---------------------------------------- def test_data_field_lists_match_emitter_exactly(): envelope = load_lambda_module("wo", "shoc_emitter/envelope") # Same fields, same order -- the receiver sees one schema whether an # event arrived live or via replay. assert list(replay.WO_DATA_FIELDS) == list(envelope.WO_DATA_FIELDS) assert list(replay.COMMENT_DATA_FIELDS) == list(envelope.COMMENT_DATA_FIELDS) assert "source_email_s3_key" not in replay.WO_DATA_FIELDS assert replay.SCHEMA_VERSION == envelope.SCHEMA_VERSION assert replay.SOURCE == envelope.SOURCE def test_replay_envelopes_carry_replay_true_and_exclude_s3_key(): state = replay.build_state_event(dict(WO_ITEM)) comment = replay.build_comment_event(dict(COMMENT_ITEM)) assert state["replay"] is True assert comment["replay"] is True # State replays are always work_order.updated: current-state rebuilds # cannot distinguish the original created/cancelled, and the receiver # upserts idempotently. assert state["event_type"] == "work_order.updated" assert comment["event_type"] == "work_order.comment_added" assert "source_email_s3_key" not in state["data"] assert set(state["data"]) == set(replay.WO_DATA_FIELDS) assert set(comment["data"]) == set(replay.COMMENT_DATA_FIELDS) def test_delivery_id_is_stable_across_builds(): first = replay.build_comment_event(dict(COMMENT_ITEM)) second = replay.build_comment_event(dict(COMMENT_ITEM)) assert first["delivery_id"] == second["delivery_id"] assert first["delivery_id"].startswith("replay-") state_first = replay.build_state_event(dict(WO_ITEM)) state_second = replay.build_state_event(dict(WO_ITEM)) assert state_first["delivery_id"] == state_second["delivery_id"] # State and comment ids never collide (different table seeds). assert first["delivery_id"] != state_first["delivery_id"] # --- --since parsing --------------------------------------------------------- def test_parse_since_canonicalizes_utc(): assert replay.parse_since("2026-07-24T02:00:00Z") == "2026-07-24T02:00:00+00:00" assert ( replay.parse_since("2026-07-24T02:00:00+00:00") == "2026-07-24T02:00:00+00:00" ) @pytest.mark.parametrize( "value", ["2026-07-24T02:00:00", "2026-07-24T02:00:00+02:00", "yesterday"], ids=["naive", "non-utc-offset", "garbage"], ) def test_parse_since_rejects_non_utc_or_garbage(value): with pytest.raises(SystemExit): replay.parse_since(value)