informed-decision/tests/test_audit_delivery.py

from datetime import datetime, timezone
import json

import pytest

from informed_decision.audit import AuditCoreSink, AuditDeliveryError, OutboxWorker
from informed_decision.http_transport import TransportError
from informed_decision.store import Store
from test_durable_store import storage, present


class Replies:
    def __init__(self, *replies):
        self.replies = list(replies)
        self.calls = []

    def request(self, method, url, **kwargs):
        self.calls.append((method, url, kwargs))
        reply = self.replies.pop(0)
        if isinstance(reply, Exception): raise reply
        return reply


@pytest.mark.parametrize("status,name", [(202, "accepted"), (200, "duplicate")])
def test_only_confirmed_receiver_receipt_marks_delivered(storage, status, name):
    store, memo = storage
    present(store, memo)
    event = store.evidence()[0]
    transport = Replies((status, {"status": name, "reference": "audit:" + event["id"]}))
    sink = AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=transport)
    assert OutboxWorker(store, sink).run_once()["delivered"] == 1
    row = Store(store.path).outbox()[0]
    assert row["state"] == "delivered" and row["receiver_reference"] == "audit:" + event["id"]
    method, url, params = transport.calls[0]
    assert (method, url) == ("POST", "https://audit.test/v1/events")
    assert params["headers"]["Idempotency-Key"] == event["id"]
    assert params["body"] == event["envelope"].encode()
    assert "synthetic-token" not in event["envelope"]


@pytest.mark.parametrize("reply,code,blocked", [
    ((400, {"error": "secret-sentinel"}), "rejected", True),
    ((401, {}), "unauthorized", True), ((403, {}), "unauthorized", True),
    ((409, {}), "conflict", True), ((503, {}), "unavailable", False),
    ((500, {}), "unavailable", False), (TransportError("secret-sentinel"), "unavailable", False),
    ((202, {"status": "accepted", "reference": "audit:other-event"}), "invalid_receipt", False),
    ((200, {"status": "ok", "reference": "audit:fixture"}), "invalid_receipt", False),
])
def test_refusals_preserve_pending_evidence_with_bounded_error(storage, reply, code, blocked):
    store, memo = storage
    present(store, memo)
    before = store.evidence()
    transport = Replies(reply)
    worker = OutboxWorker(store, AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=transport))
    result = worker.run_once()
    assert result["blocked" if blocked else "retrying"] == 1
    row = Store(store.path).outbox()[0]
    assert row["state"] == ("blocked" if blocked else "pending")
    assert row["last_error"] == code and "secret-sentinel" not in str(row)
    assert store.evidence() == before
    assert worker.run_once() == {"delivered": 0, "retrying": 0, "blocked": 0}


def test_repaired_credential_requeues_original_bytes(storage):
    store, memo = storage
    present(store, memo)
    event = store.evidence()[0]
    transport = Replies((401, {}), (202, {"status": "accepted", "reference": "audit:" + event["id"]}))
    tokens = iter(["old-synthetic", "new-synthetic"])
    worker = OutboxWorker(store, AuditCoreSink("https://audit.test", lambda: next(tokens), transport=transport))
    assert worker.run_once()["blocked"] == 1
    store.requeue_blocked(event["id"])
    assert worker.run_once()["delivered"] == 1
    assert transport.calls[0][2]["body"] == transport.calls[1][2]["body"]
    assert transport.calls[1][2]["headers"]["Authorization"] == "Bearer new-synthetic"


def test_unavailable_policy_record_is_never_a_human_decline(storage):
    store, memo = storage
    store.record_unreachable(memo.id, "access-engine")
    envelope = json.loads(store.evidence()[0]["envelope"])
    assert envelope["type"] == "informed-decision.stance_application"
    assert envelope["data"]["stance_applied"] == "fail_closed"
    assert envelope["data"]["decision_attributable"] is False
    assert "verb" not in envelope["data"]


def test_reconciliation_reports_gap_without_claiming_completeness(storage):
    store, memo = storage
    present(store, memo)
    since, until = "2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z"
    remote = {"source": "informed-decision", "tenant": "tenant:platform", "since": since, "until": until, "counts": []}
    transport = Replies((200, remote))
    worker = OutboxWorker(store, AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=transport))
    report = worker.reconcile(since, until)
    assert report["count_values_match"] is False
    assert report["counts"]["informed-decision.presentation"] == {"source": 1, "receiver": 0}
    assert report["completeness_proven"] is report["reconstructability_proven"] is False
    assert report["source_time_basis"] == "occurred_at" and report["receiver_time_basis"] == "accepted_at"
    assert report["automatic_loss_finding"] is False
    with pytest.raises(ValueError): worker.reconcile(None, until)


@pytest.mark.parametrize("change", [{"source": "other-source"}, {"tenant": "other-tenant"},
    {"since": "2019-01-01T00:00:00Z"}, {"counts": [{"class": "presentation", "count": True}]},
    {"counts": [{"class": "presentation", "count": -1}]},
    {"counts": [{"class": "presentation", "count": 1}, {"class": "presentation", "count": 1}]}])
def test_reconciliation_rejects_wrong_scope_window_or_shape(change):
    body = {"source": "informed-decision", "tenant": "tenant:platform", "since": "2020-01-01T00:00:00Z",
            "until": "2100-01-01T00:00:00Z", "counts": [], **change}
    sink = AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=Replies((200, body)))
    with pytest.raises(AuditDeliveryError): sink.counts("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")