informed-decision/tests/test_durable_component.py

"""Actual Approval Engine and Audit Core APIs, with synthetic identities/custody."""

import io
import json
import os
from pathlib import Path
import sys
from urllib.parse import urlsplit

import pytest

from informed_decision.audit import AuditCoreSink, AuditDeliveryError, OutboxWorker
from informed_decision.disposition import Actor, ActorKind, Verb
from informed_decision.http_transport import TransportError
from informed_decision.memo import PacketItem
from informed_decision.store import Store, Conflict
from test_approval_component import component, signing_key
from test_skeleton import make_memo
from test_durable_store import storage, present


@pytest.fixture
def receiver(tmp_path):
    source = os.environ.get("INFD_AUDIT_CORE_SOURCE")
    if not source:
        pytest.skip("set INFD_AUDIT_CORE_SOURCE for actual Audit Core contract checks")
    assert (Path(source) / "audit_core/ingestion.py").is_file()
    sys.path.insert(0, source)
    from audit_core.ingestion import IngestionApplication
    from audit_core.senders import SenderIdentity, SenderRegistry
    from audit_core.sqlite_backend import SQLiteAuditBackend
    backend = SQLiteAuditBackend(str(tmp_path / "independent-receiver.sqlite"))
    registry = SenderRegistry([SenderIdentity(name="informed-decision", tokens=("synthetic-audit-token",),
        sources=frozenset({"informed-decision"}), tenants=frozenset({"tenant:platform"}),
        evidence_kind="load-bearing", secret_policy="redact", may_read=False),
        SenderIdentity(name="independent-fixture-reader", tokens=("synthetic-auditor-token",),
            sources=frozenset({"informed-decision"}), tenants=frozenset({"tenant:platform"}),
            may_read=True, may_write=False)])
    app = IngestionApplication(backend, registry)

    class Transport:
        lose_reply = False
        def request(self, method, url, *, headers=None, body=None):
            parsed = urlsplit(url)
            environ = {"PATH_INFO": parsed.path, "QUERY_STRING": parsed.query, "REQUEST_METHOD": method,
                "CONTENT_LENGTH": str(len(body or b"")), "wsgi.input": io.BytesIO(body or b""),
                "HTTP_AUTHORIZATION": headers.get("Authorization", ""),
                "HTTP_IDEMPOTENCY_KEY": headers.get("Idempotency-Key", "")}
            result = {}
            out = b"".join(app(environ, lambda status, headers: result.update(status=int(status.split()[0]))))
            if self.lose_reply:
                self.lose_reply = False
                raise TransportError("simulated lost receiver response")
            return result["status"], json.loads(out)

    transport = Transport()
    sink = AuditCoreSink("https://audit.test", lambda: "synthetic-audit-token", transport=transport)
    return sink, transport, backend


def durable_setup(tmp_path, component):
    client, engine, transport, session, _ = component
    private = tmp_path / "holder"
    private.mkdir(mode=0o700)
    store = Store(private / "review.sqlite")
    digest = store.put_document(b"synthetic factory review packet")
    approval = client.get_approval("fixture")
    memo = make_memo(approval_id="fixture", approval_binding_digest=approval["binding"]["digest"],
        packet=(PacketItem("doc-1", "Factory fixture", digest),))
    store.save_memo(memo)
    p = store.present(memo.id, principal_sub=session.subject, tenant=session.tenant, principal_type=session.principal_type)
    actor = Actor(session.subject, ActorKind.PERSON)
    store.acknowledge(p.id, actor, ["h-1"])
    return store, p, actor


def test_actual_entry_and_independent_receiver_survive_holder_restart(component, receiver, tmp_path):
    client, engine, _, _, _ = component
    sink, receiver_transport, backend = receiver
    store, p, actor = durable_setup(tmp_path, component)
    d = store.record_disposition(p.id, actor, Verb.ACCEPT, operation_id="synthetic-click")
    attempt = store.begin_submission(d.id)  # Harness only; native policy admission is not exercised.
    result = client.add_entry("fixture")
    assert store.finish_submission(d.id, attempt, result) == "confirmed"
    worker = OutboxWorker(Store(store.path), sink)
    assert worker.run_once()["delivered"] == 4
    assert worker.reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")["count_values_match"]
    recovered = Store(store.path).presentation_for_entry(*result.correlation)
    assert recovered[0].id == d.id and recovered[1].id == p.id
    assert recovered[3] == {"doc-1": b"synthetic factory review packet"}
    assert len(engine.get("fixture").entries) == 1
    # This sender can count its stream without gaining archive read privileges.
    with pytest.raises(AuditDeliveryError, match="unauthorized"):
        sink._request("GET", "/v1/events")
    event = store.evidence()[0]
    status, body = receiver_transport.request("GET", "https://audit.test/v1/events/" + event["id"],
        headers={"Authorization": "Bearer synthetic-auditor-token"})
    assert status == 200 and body["details"]["data"]["view_hash"] == p.view_hash
    assert "synthetic factory review packet" not in json.dumps(body)
    assert not any("synthetic factory review packet" in row["envelope"] for row in store.evidence())


def test_lost_receiver_reply_reuses_exact_event_and_deduplicates(component, receiver, tmp_path):
    sink, transport, backend = receiver
    store, _, _ = durable_setup(tmp_path, component)
    first = store.claim_delivery(now=10)
    transport.lose_reply = True
    with pytest.raises(AuditDeliveryError, match="unavailable"): sink.deliver(first[0], first[2])
    # Simulate crash before marking local delivery; its lease expires after restart.
    replay = Store(store.path).claim_delivery(now=41)
    assert replay[0] == first[0] and replay[2] == first[2]
    reference = sink.deliver(replay[0], replay[2])
    store.finish_delivery(replay[0], replay[1], reference=reference)
    OutboxWorker(store, sink).run_once()
    report = OutboxWorker(store, sink).reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")
    assert report["count_values_match"] and report["counts"]["informed-decision.presentation"] == {"source": 2, "receiver": 2}


def test_lost_engine_reply_does_not_rebind_or_reassign_presentation(component, tmp_path):
    client, engine, transport, _, _ = component
    store, p, actor = durable_setup(tmp_path, component)
    d = store.record_disposition(p.id, actor, Verb.ACCEPT, operation_id="synthetic-click")
    attempt = store.begin_submission(d.id)
    client.add_entry("fixture")  # Real API commits; the caller's response is deliberately discarded.
    assert store.finish_submission(d.id, attempt) == "unresolved"
    calls = len(transport.calls)
    with pytest.raises(Conflict): Store(store.path).begin_submission(d.id)
    assert len(transport.calls) == calls and len(engine.get("fixture").entries) == 1
    assert store.submission(d.id)["approved_at"] is None


def test_delayed_acceptance_has_a_different_window_without_being_lost(storage, receiver, monkeypatch):
    from informed_decision import evidence, presentation
    monkeypatch.setattr(evidence, "_now", lambda: "2020-06-01T12:00:00Z")
    monkeypatch.setattr(presentation, "_now", lambda: "2020-06-01T12:00:00Z")
    store, memo = storage
    present(store, memo)
    sink, _, backend = receiver
    worker = OutboxWorker(store, sink)
    assert worker.run_once()["delivered"] == 1
    historical = worker.reconcile("2020-01-01T00:00:00Z", "2021-01-01T00:00:00Z")
    assert not historical["count_values_match"] and not historical["automatic_loss_finding"]
    assert worker.reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")["count_values_match"]