audit-core/tests/test_independent_reconciliation.py

"""AUDIT-WP-0013-T06 — independent cross-source reconciliation counts.

NK-WP-0035-T04: an operator reader registered for one source must be able to
count another source's window, without payloads, and only with full tenant
scope. A writer still may not count another source.
"""

import io
import json

import pytest

from audit_core.ingestion import IngestionApplication
from audit_core.senders import SenderIdentity, SenderRegistry
from audit_core.sqlite_backend import SQLiteAuditBackend


def invoke(app, path, *, token, method="GET", payload=None, key=None):
    raw = json.dumps(payload).encode() if payload is not None else b""
    environ = {
        "PATH_INFO": path.split("?")[0],
        "QUERY_STRING": path.split("?")[1] if "?" in path else "",
        "REQUEST_METHOD": method,
        "CONTENT_LENGTH": str(len(raw)),
        "wsgi.input": io.BytesIO(raw),
        "HTTP_AUTHORIZATION": f"Bearer {token}",
    }
    if key:
        environ["HTTP_IDEMPOTENCY_KEY"] = key
    result = {}
    out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
    return result["status"], (json.loads(out) if out else {})


TENANT = "tenant:friendly:binky"
INSTANCE = "b351fa6b-e0d6-4f88-b7c9-6bacb1a112b7"
WINDOW = "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00"

QONTO = SenderIdentity(
    name="qonto-assistant", tokens=("qonto",), sources=frozenset({"qonto-assistant"}),
    tenants=frozenset({TENANT}), may_write=True, may_read=False,
)
USER_ENGINE = SenderIdentity(
    name="user-engine", tokens=("writer",), sources=frozenset({"user-engine"}),
    tenants=frozenset({"*"}), may_write=True, may_read=False,
)
OPERATOR = SenderIdentity(
    name="operator", tokens=("operator",), sources=frozenset({"user-engine"}),
    tenants=frozenset({"*"}), may_write=False, may_read=True,
)
SCOPED_READER = SenderIdentity(
    name="scoped-reader", tokens=("scoped",), sources=frozenset({"user-engine"}),
    tenants=frozenset({TENANT}), may_write=False, may_read=True,
)


@pytest.fixture
def app(tmp_path):
    backend = SQLiteAuditBackend(str(tmp_path / "s.db"))
    return IngestionApplication(
        backend, SenderRegistry([QONTO, USER_ENGINE, OPERATOR, SCOPED_READER])
    )


def emit(app, event_id, action, *, instance=INSTANCE, secret=None):
    data = {"stream_instance_id": instance, "sequence": 1}
    if secret:
        data["binding"] = secret
    payload = {
        "id": event_id, "type": action, "source": "qonto-assistant",
        "subject": "s:1", "tenant": TENANT, "correlation_id": "c-1",
        "occurred_at": "2026-10-15T19:40:00+00:00", "data": data,
    }
    status, _ = invoke(app, "/v1/events", token="qonto", method="POST",
                       payload=payload, key=event_id)
    assert status.startswith("202")


def count(app, token, query):
    return invoke(app, "/v1/reconciliation?" + query + WINDOW, token=token)


def test_full_scope_reader_counts_another_source(app):
    emit(app, "h-1", "audit.heartbeat")
    emit(app, "h-2", "audit.heartbeat")
    emit(app, "d-1", "audit.deny")
    emit(app, "x-1", "audit.deny", instance="other-instance")
    status, body = count(
        app, "operator",
        f"source=qonto-assistant&tenant={TENANT}&stream_instance_id={INSTANCE}",
    )
    assert status.startswith("200")
    assert body["counts"] == [
        {"class": "audit.deny", "count": 1},
        {"class": "audit.heartbeat", "count": 2},
    ]
    assert body["basis"] == "independent_reader"
    assert body["stream_instance_id"] == INSTANCE


def test_full_scope_reader_may_omit_tenant(app):
    emit(app, "d-1", "audit.deny")
    status, body = count(app, "operator", "source=qonto-assistant")
    assert status.startswith("200")
    assert body["tenant"] is None
    assert body["counts"] == [{"class": "audit.deny", "count": 1}]


def test_independent_counts_never_carry_payloads(app):
    emit(app, "d-1", "audit.deny", secret="value-not-to-leak")
    _, body = count(app, "operator", f"source=qonto-assistant&tenant={TENANT}")
    assert set(body) == {
        "source", "tenant", "since", "until", "counts", "means", "basis",
    }
    assert "value-not-to-leak" not in json.dumps(body)


def test_tenant_scoped_reader_may_not_count_another_source(app):
    emit(app, "d-1", "audit.deny")
    status, body = count(app, "scoped", f"source=qonto-assistant&tenant={TENANT}")
    assert status.startswith("403")
    assert body["error"] == "full_tenant_scope_required"


def test_writer_still_may_not_count_another_source(app):
    status, body = count(app, "writer", f"source=qonto-assistant&tenant={TENANT}")
    assert status.startswith("403")
    assert body["error"] == "source_not_allowed"


def test_own_source_reconciliation_is_unchanged(app):
    emit(app, "d-1", "audit.deny")
    status, body = count(app, "qonto", f"source=qonto-assistant&tenant={TENANT}")
    assert status.startswith("200")
    assert "basis" not in body
    assert body["counts"] == [{"class": "audit.deny", "count": 1}]


def test_independent_reader_still_needs_a_bounded_window(app):
    status, body = invoke(
        app, "/v1/reconciliation?source=qonto-assistant", token="operator"
    )
    assert status.startswith("400")
    assert body["error"] == "since_and_until_required"