#!/usr/bin/env python3 """Synthetic, offline acceptance fixture for an official webhook replacement. No network requests, credentials, or external services. It deliberately does not infer absent events or reconstruct transient historical state. """ from dataclasses import dataclass from enum import Enum from typing import Dict, Iterable, Optional, Tuple class Decision(str, Enum): USE_WEBHOOK_WITH_RECONCILIATION = "USE_WEBHOOK_WITH_RECONCILIATION" KEEP_APPROVED_POLLING = "KEEP_APPROVED_POLLING" HOLD = "HOLD" @dataclass(frozen=True) class Event: delivery_id: Optional[str] object_id: str source_version: int state: str @dataclass(frozen=True) class Assessment: decision: Decision reason: str materialized: Dict[str, Tuple[int, str]] def assess(*, coverage_proven: bool, signature_proven: bool, retry_window_closed: bool, reconciliation_available: bool, every_event_required: bool, approved_poll_acceptable: bool, events: Iterable[Event], authoritative: Dict[str, Tuple[int, str]]) -> Assessment: """Assess synthetic webhook evidence without treating silence as no change.""" gates = (coverage_proven, signature_proven, retry_window_closed, reconciliation_available, every_event_required, approved_poll_acceptable) if not all(isinstance(value, bool) for value in gates): return Assessment(Decision.HOLD, "A required acceptance gate is unknown rather than a boolean decision.", {}) if every_event_required: return Assessment(Decision.HOLD, "Every-event completeness is required; current-state routes cannot prove it. Escalate an auditable-feed redesign.", {}) if not coverage_proven or not signature_proven or not reconciliation_available: if approved_poll_acceptable: return Assessment(Decision.KEEP_APPROVED_POLLING, "Webhook evidence is insufficient; retain the independently accepted current-state polling baseline.", {}) return Assessment(Decision.HOLD, "Coverage, signature, or authoritative reconciliation is not proven.", {}) if not retry_window_closed: return Assessment(Decision.HOLD, "The declared retry window is still open; reconciliation is premature.", {}) if not authoritative: return Assessment(Decision.HOLD, "No nonempty authoritative scope was supplied for reconciliation.", {}) for object_id, value in authoritative.items(): if not isinstance(object_id, str) or not object_id.strip() or not isinstance(value, tuple) or len(value) != 2: return Assessment(Decision.HOLD, "Authoritative scope is malformed.", {}) version, state = value if isinstance(version, bool) or not isinstance(version, int) or version < 0 or not isinstance(state, str) or not state.strip(): return Assessment(Decision.HOLD, "Authoritative version/state is invalid.", {}) event_list = list(events) if not event_list: return Assessment(Decision.HOLD, "No observed deliveries; silence is not evidence of no change.", {}) seen: Dict[str, Event] = {} version_states: Dict[Tuple[str, int], str] = {} materialized: Dict[str, Tuple[int, str]] = {} for event in event_list: if not isinstance(event.object_id, str) or not event.object_id.strip() or isinstance(event.source_version, bool) or not isinstance(event.source_version, int) or event.source_version < 0 or not isinstance(event.state, str) or not event.state.strip(): return Assessment(Decision.HOLD, "Delivery has an invalid object, version, or state.", materialized) if not isinstance(event.delivery_id, str) or not event.delivery_id.strip(): return Assessment(Decision.HOLD, "A delivery lacks a stable ID; safe deduplication is impossible.", materialized) previous_delivery = seen.get(event.delivery_id) if previous_delivery is not None: if previous_delivery != event: return Assessment(Decision.HOLD, "One delivery ID has conflicting content; quarantine the collision.", materialized) continue seen[event.delivery_id] = event key = (event.object_id, event.source_version) if key in version_states and version_states[key] != event.state: return Assessment(Decision.HOLD, "One object/version has conflicting states; do not choose by arrival order.", materialized) version_states[key] = event.state old = materialized.get(event.object_id) # Arrival order is deliberately ignored; the provider version wins. if old is None or event.source_version > old[0]: materialized[event.object_id] = (event.source_version, event.state) if set(materialized) != set(authoritative): return Assessment(Decision.HOLD, "Reconciliation scope differs from observed scope; do not infer missing objects.", materialized) if materialized != authoritative: return Assessment(Decision.HOLD, "Current authoritative state differs; investigate and reconcile before switching.", materialized) return Assessment(Decision.USE_WEBHOOK_WITH_RECONCILIATION, "Synthetic current-state evidence matches after the retry window; keep bounded reconciliation.", materialized) def expect(name: str, got: Assessment, wanted: Decision) -> None: assert got.decision == wanted, f"{name}: expected {wanted}, got {got.decision}: {got.reason}" print(f"PASS {name}: {got.decision.value} — {got.reason}") def main() -> None: final = {"offer-7": (3, "public-price-95")} ordered = [Event("d-1", "offer-7", 1, "public-price-100"), Event("d-2", "offer-7", 3, "public-price-95")] common = dict(coverage_proven=True, signature_proven=True, retry_window_closed=True, reconciliation_available=True, every_event_required=False, approved_poll_acceptable=True, authoritative=final) expect("coverage-dedup-and-current-state", assess(**common, events=ordered + [ordered[1]]), Decision.USE_WEBHOOK_WITH_RECONCILIATION) expect("out-of-order-arrival", assess(**common, events=list(reversed(ordered))), Decision.USE_WEBHOOK_WITH_RECONCILIATION) expect("reconciliation-mismatch", assess(**common, events=[ordered[0]]), Decision.HOLD) expect("retry-window-open", assess(**{**common, "retry_window_closed": False}, events=ordered), Decision.HOLD) expect("missing-delivery-id", assess(**common, events=[Event(None, "offer-7", 3, "public-price-95")]), Decision.HOLD) expect("every-event-requirement", assess(**{**common, "every_event_required": True}, events=ordered), Decision.HOLD) expect("accepted-baseline-when-webhook-ineligible", assess(**{**common, "reconciliation_available": False}, events=ordered), Decision.KEEP_APPROVED_POLLING) conflict = [Event("d-3", "offer-7", 3, "public-price-95"), Event("d-4", "offer-7", 3, "public-price-90")] expect("same-version-conflict-forward", assess(**common, events=conflict), Decision.HOLD) expect("same-version-conflict-reversed", assess(**common, events=list(reversed(conflict))), Decision.HOLD) expect("delivery-id-collision", assess(**common, events=[Event("d-5", "offer-7", 3, "public-price-95"), Event("d-5", "offer-7", 4, "public-price-90")]), Decision.HOLD) expect("empty-scope", assess(**{**common, "authoritative": {}}, events=[]), Decision.HOLD) expect("unknown-every-event-gate", assess(**{**common, "every_event_required": None}, events=ordered), Decision.HOLD) expect("invalid-boolean-version", assess(**common, events=[Event("d-6", "offer-7", True, "public-price-95")]), Decision.HOLD) older_conflict = [ordered[1], ordered[0], Event("d-7", "offer-7", 1, "public-price-101")] expect("older-version-conflict-after-newer", assess(**common, events=older_conflict), Decision.HOLD) expect("older-version-conflict-before-newer", assess(**common, events=list(reversed(older_conflict))), Decision.HOLD) print("All cases are synthetic. A matching current state does not prove that no transient change occurred.") if __name__ == "__main__": main()