diff --git a/src/adcp/reporting/adjustment_evidence.py b/src/adcp/reporting/adjustment_evidence.py new file mode 100644 index 000000000..406e9262e --- /dev/null +++ b/src/adcp/reporting/adjustment_evidence.py @@ -0,0 +1,494 @@ +"""Buyer adjustment evidence and deterministic receipt construction. + +These primitives do not authenticate a transport, select a revision/history leaf, +or persist submission intents. The caller supplies trusted scope and a selected +official revision from a complete authorized history. Capture the raw adjustment +before domain-model normalization; a model dump is never raw evidence. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from decimal import Decimal +from typing import Any, Literal, NoReturn + +import rfc8785 + +from adcp.reporting.evidence import ( + aware_utc, + consumer_reference, + principal_reference, + reporting_identifier, +) +from adcp.types import ( + ReportingAdjustment, + ReportingAdjustmentReceipt, + ReportingObligation, + ReportingRevision, +) +from adcp.validation.schema_loader import get_named_validator + +__all__ = [ + "ReportingAdjustmentEvidence", + "ReportingAdjustmentEvidenceError", + "ReportingAdjustmentEvidenceLimits", + "ReportingAdjustmentReceiptContext", + "ReportingAdjustmentScope", + "build_reporting_adjustment_receipt", + "capture_reporting_adjustment_evidence", +] + +ErrorCode = Literal[ + "INVALID_EVIDENCE", + "EVIDENCE_LIMIT_EXCEEDED", + "TYPED_EVIDENCE_MISMATCH", + "INVALID_CONTEXT", + "ADJUSTMENT_CONTEXT_MISMATCH", + "INVALID_RECEIPT", + "RECEIPT_TERMINAL", + "SCHEMA_UNAVAILABLE", +] + + +class ReportingAdjustmentEvidenceError(ValueError): + """Closed diagnostics; no input, decoder error or validation context.""" + + def __init__(self, code: ErrorCode) -> None: + self.code = code + super().__init__(code) + + +def _fail(code: ErrorCode) -> NoReturn: + raise ReportingAdjustmentEvidenceError(code) from None + + +@dataclass(frozen=True, slots=True) +class ReportingAdjustmentScope: + """Caller-asserted trusted identities, never derived from response fields. + + Constructing this value does not establish authentication or authorization. + Resolve aliases before capture; these identities are preserved verbatim. + """ + + seller_identity: str = field(repr=False) + account_id: str = field(repr=False) + consumer_id: str = field(repr=False) + reporting_obligation_id: str = field(repr=False) + + def __post_init__(self) -> None: + valid = False + try: + consumer_reference(self.seller_identity) + principal_reference(self.account_id) + consumer_reference(self.consumer_id) + reporting_identifier(self.reporting_obligation_id) + valid = True + except (ValueError, TypeError): + # Normalize parse errors after leaving the exception handler. + valid = False + if not valid: + _fail("INVALID_CONTEXT") + + +@dataclass(frozen=True, slots=True) +class ReportingAdjustmentEvidenceLimits: + """Caller-selected admission bounds, checked before schema/model work. + + Defaults are conservative; there is no separate SDK-wide ceiling on trusted + caller configuration. Raising these limits expands the admitted workload. + """ + + max_bytes: int = 65_536 + max_depth: int = 12 + max_nodes: int = 4096 + + def __post_init__(self) -> None: + if any( + type(v) is not int or v < 1 for v in (self.max_bytes, self.max_depth, self.max_nodes) + ): + _fail("INVALID_EVIDENCE") + + +def _bounded(value: object, limits: ReportingAdjustmentEvidenceLimits) -> None: + pending = [(value, 0)] + seen: set[int] = set() + nodes = size = 0 + while pending: + item, depth = pending.pop() + nodes += 1 + if nodes > limits.max_nodes or depth > limits.max_depth: + _fail("EVIDENCE_LIMIT_EXCEEDED") + if type(item) in (dict, list): + if id(item) in seen: + _fail("INVALID_EVIDENCE") + seen.add(id(item)) + size += 2 + if type(item) is dict: + if len(item) > limits.max_nodes: + _fail("EVIDENCE_LIMIT_EXCEEDED") + for key, child in item.items(): + if type(key) is not str: + _fail("INVALID_EVIDENCE") + pending.extend(((key, depth + 1), (child, depth + 1))) + elif type(item) is list: + if len(item) > limits.max_nodes: + _fail("EVIDENCE_LIMIT_EXCEEDED") + pending.extend((child, depth + 1) for child in item) + elif type(item) is str: + if len(item) > limits.max_bytes: + _fail("EVIDENCE_LIMIT_EXCEEDED") + size += len(item.encode("utf-8")) + elif item is None or type(item) is bool: + size += 5 + elif type(item) is int: + if abs(item) > 9_007_199_254_740_991: + _fail("INVALID_EVIDENCE") + size += 20 + elif type(item) is float: + if not math.isfinite(item): + _fail("INVALID_EVIDENCE") + size += 24 + else: + _fail("INVALID_EVIDENCE") + if size > limits.max_bytes: + _fail("EVIDENCE_LIMIT_EXCEEDED") + + +def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + _fail("INVALID_EVIDENCE") + result[key] = value + return result + + +def _nonfinite(_value: str) -> NoReturn: + _fail("INVALID_EVIDENCE") + + +def _instant(parsed: datetime, raw: str | None = None) -> tuple[datetime, Decimal]: + """Compare RFC3339 instants without losing sub-microsecond wire precision.""" + fraction = re.search(r"\.(\d+)", raw) if raw is not None else None + digits = fraction.group(1) if fraction else f"{parsed.microsecond:06d}" + return aware_utc(parsed).replace(microsecond=0), Decimal("0." + digits) + + +def _schema_valid(value: dict[str, Any], name: str) -> bool: + valid: bool | None = None + try: + validator = get_named_validator("core/" + name + ".json") + if validator is not None: + valid = bool(validator.is_valid(value)) + except Exception: + # Resolver errors can include evidence values. Expose only a closed code. + valid = None + if valid is None: + _fail("SCHEMA_UNAVAILABLE") + return valid + + +@dataclass(frozen=True, slots=True) +class ReportingAdjustmentEvidence: + """Immutable captured JSON; explicit access can expose private evidence. + + Prefer capture_reporting_adjustment_evidence. Construction revalidates the + input, so serialized evidence may be restored with the same checks. ``bytes`` + records strict duplicate-key admission; ``mapping`` cannot prove what its + upstream decoder discarded. Neither value proves transport authenticity. + """ + + scope: ReportingAdjustmentScope = field(repr=False) + raw_json: bytes = field(repr=False) + input_kind: Literal["bytes", "mapping"] + limits: ReportingAdjustmentEvidenceLimits = field( + default_factory=ReportingAdjustmentEvidenceLimits, repr=False + ) + canonical_json: bytes = field(init=False, repr=False) + observed_adjustment_sha256: str = field(init=False) + + def __post_init__(self) -> None: + if ( + type(self.scope) is not ReportingAdjustmentScope + or type(self.limits) is not ReportingAdjustmentEvidenceLimits + or type(self.raw_json) is not bytes + or self.input_kind not in ("bytes", "mapping") + ): + _fail("INVALID_EVIDENCE") + if len(self.raw_json) > self.limits.max_bytes: + _fail("EVIDENCE_LIMIT_EXCEEDED") + valid = False + canonical = b"" + try: + raw = json.loads( + self.raw_json.decode("utf-8"), object_pairs_hook=_pairs, parse_constant=_nonfinite + ) + _bounded(raw, self.limits) + if type(raw) is dict and _schema_valid(raw, "reporting-adjustment"): + typed = ReportingAdjustment.model_validate(raw) + names = [item.name for item in typed.control_total_deltas] + valid = ( + typed.canonical_adjustment_sha256 is not None + and len(names) == len(set(names)) + and _instant(typed.accounting_period.start, raw["accounting_period"]["start"]) + < _instant(typed.accounting_period.end, raw["accounting_period"]["end"]) + and _instant(typed.correction_observed_at, raw["correction_observed_at"]) + <= _instant(typed.created_at, raw["created_at"]) + ) + if valid: + canonical = rfc8785.dumps( + { + key: value + for key, value in raw.items() + if key != "canonical_adjustment_sha256" + } + ) + except ReportingAdjustmentEvidenceError: + raise + except (ValueError, TypeError, OverflowError, RecursionError): + # Do not retain a decoder exception in the public error context. + valid = False + if not valid: + _fail("INVALID_EVIDENCE") + object.__setattr__(self, "canonical_json", canonical) + object.__setattr__( + self, "observed_adjustment_sha256", hashlib.sha256(canonical).hexdigest() + ) + + @property + def adjustment(self) -> ReportingAdjustment: + """Fresh mutable model; mutations cannot alter retained evidence.""" + return ReportingAdjustment.model_validate_json(self.raw_json) + + +def capture_reporting_adjustment_evidence( + raw: bytes | Mapping[str, Any], + *, + typed_adjustment: ReportingAdjustment, + scope: ReportingAdjustmentScope, + limits: ReportingAdjustmentEvidenceLimits = ReportingAdjustmentEvidenceLimits(), +) -> ReportingAdjustmentEvidence: + """Capture authenticated-ingress data supplied by the caller, then align views. + + Bytes retain the supplied spelling and reject duplicate keys. A mapping is + only evidence of those supplied decoded values, not of original wire bytes, + duplicate-key absence, or numeric information lost by its upstream decoder. + Never supply a model dump or debug capture as the mapping. + """ + if type(limits) is not ReportingAdjustmentEvidenceLimits: + _fail("INVALID_EVIDENCE") + encoded: bytes | None = None + kind: Literal["bytes", "mapping"] = "bytes" + try: + if type(raw) is bytes: + encoded = raw + elif isinstance(raw, Mapping): + kind = "mapping" + if len(raw) > limits.max_nodes: + _fail("EVIDENCE_LIMIT_EXCEEDED") + value = dict(raw) + _bounded(value, limits) + encoded = json.dumps(value, ensure_ascii=False, allow_nan=False).encode("utf-8") + except ReportingAdjustmentEvidenceError: + raise + except (ValueError, TypeError, OverflowError, RecursionError): + # Normalize malformed mappings after the exception is cleared. + encoded = None + if encoded is None: + _fail("INVALID_EVIDENCE") + evidence = ReportingAdjustmentEvidence(scope, encoded, kind, limits) + aligned = False + try: + if type(typed_adjustment) is ReportingAdjustment: + supplied = ReportingAdjustment.model_validate( + typed_adjustment.model_dump(mode="json", warnings="error") + ) + aligned = supplied == evidence.adjustment + except (ValueError, TypeError, OverflowError, RecursionError): + # Keep model diagnostics out of the public mismatch error. + aligned = False + if not aligned: + _fail("TYPED_EVIDENCE_MISMATCH") + return evidence + + +@dataclass(frozen=True, slots=True) +class ReportingAdjustmentReceiptContext: + """Caller-asserted selected official and exact ownership, not a selector. + + The caller must establish complete authorized history and the current leaf + before using this primitive. from_selection validates its supplied binding; + it cannot establish completeness or authorization from those objects alone. + """ + + scope: ReportingAdjustmentScope = field(repr=False) + reporting_revision_id: str = field(repr=False) + finalized_at: datetime = field(repr=False) + control_total_units: tuple[tuple[str, str | None], ...] = field(repr=False) + + def __post_init__(self) -> None: + valid = False + try: + if type(self.scope) is ReportingAdjustmentScope: + reporting_identifier(self.reporting_revision_id) + final = aware_utc(self.finalized_at) + if type(self.control_total_units) not in (tuple, list) or any( + type(item) not in (tuple, list) for item in self.control_total_units + ): + _fail("INVALID_CONTEXT") + units = tuple(tuple(item) for item in self.control_total_units) + valid = bool(units) and all( + len(item) == 2 + and type(item[0]) is str + and (item[1] is None or type(item[1]) is str) + for item in units + ) + valid = valid and len({item[0] for item in units}) == len(units) + if valid: + object.__setattr__(self, "finalized_at", final) + object.__setattr__(self, "control_total_units", units) + except (ValueError, TypeError, AttributeError): + # Normalize parse errors after leaving the exception handler. + valid = False + if not valid: + _fail("INVALID_CONTEXT") + + @classmethod + def from_selection( + cls, + scope: ReportingAdjustmentScope, + *, + obligation: ReportingObligation, + revision: ReportingRevision, + revision_owner: str, + ) -> ReportingAdjustmentReceiptContext: + """Use an explicit ownership binding, never infer identity from scope.""" + valid = False + try: + obligation = ReportingObligation.model_validate( + obligation.model_dump(mode="json", warnings="error") + ) + revision = ReportingRevision.model_validate( + revision.model_dump(mode="json", warnings="error") + ) + valid = ( + scope.reporting_obligation_id + == revision_owner + == obligation.reporting_obligation_id + and scope.account_id == obligation.account_id == revision.account_id + and str(obligation.feed_purpose) == "billing" + and str(obligation.reconciliation_mode) == "consumer_receipt" + and str(obligation.required_finality) == str(revision.finality) == "official" + and revision.finalized_at is not None + and revision.report_definition_id == obligation.report_definition_id + and revision.reporting_profile == obligation.reporting_profile + and revision.period.start == obligation.period.start + and revision.period.end == obligation.period.end + and revision.period.source_timezone == obligation.period.source_timezone + ) + except (ValueError, TypeError, AttributeError): + # Do not attach model diagnostics to the public context error. + valid = False + if not valid or revision.finalized_at is None: + _fail("INVALID_CONTEXT") + return cls( + scope, + revision.reporting_revision_id, + revision.finalized_at, + tuple((item.name, item.unit) for item in revision.control_totals), + ) + + +def build_reporting_adjustment_receipt( + evidence: ReportingAdjustmentEvidence, + context: ReportingAdjustmentReceiptContext, + *, + reporting_receipt_id: str, + observed_at: datetime, + current_receipt: ReportingAdjustmentReceipt | None = None, + rejection_codes: tuple[str, ...] = (), +) -> ReportingAdjustmentReceipt: + """Build one receipt; caller owns history proof, stable identity and persistence. + + current_receipt must be the already-verified current leaf, including for the + same trusted consumer. A lone receipt cannot prove a complete chain. Missing + current_receipt asserts no prior leaf; this function never discovers one. + The caller also owns pinned calendar/report-definition semantic checks; + report disagreement with ADJUSTMENT_SEMANTIC_MISMATCH. These primitives + carry accounting evidence, never invoice, settlement or booking authority. + """ + if ( + type(evidence) is not ReportingAdjustmentEvidence + or type(context) is not ReportingAdjustmentReceiptContext + ): + _fail("INVALID_CONTEXT") + adjustment = evidence.adjustment + raw = json.loads(evidence.raw_json) + units = dict(context.control_total_units) + if ( + evidence.scope != context.scope + or adjustment.adjusts_reporting_revision_id != context.reporting_revision_id + or _instant(adjustment.correction_observed_at, raw["correction_observed_at"]) + < _instant(context.finalized_at) + or any( + item.name not in units or item.unit != units[item.name] + for item in adjustment.control_total_deltas + ) + ): + _fail("ADJUSTMENT_CONTEXT_MISMATCH") + allowed = {"ADJUSTMENT_DIGEST_MISMATCH", "ADJUSTMENT_SEMANTIC_MISMATCH"} + if type(rejection_codes) is not tuple or any( + type(code) is not str or code not in allowed for code in rejection_codes + ): + _fail("INVALID_RECEIPT") + failures = set(rejection_codes) + if evidence.observed_adjustment_sha256 != str(adjustment.canonical_adjustment_sha256).lower(): + failures.add("ADJUSTMENT_DIGEST_MISMATCH") + payload: dict[str, Any] = { + "reporting_receipt_id": reporting_receipt_id, + "reporting_adjustment_id": adjustment.reporting_adjustment_id, + "adjusts_reporting_revision_id": adjustment.adjusts_reporting_revision_id, + "status": "rejected" if failures else "accepted", + "observed_adjustment_sha256": evidence.observed_adjustment_sha256, + } + receipt: ReportingAdjustmentReceipt | None = None + try: + reporting_identifier(reporting_receipt_id) + moment = aware_utc(observed_at) + if _instant(moment) < _instant(adjustment.created_at, raw["created_at"]): + _fail("INVALID_RECEIPT") + payload["observed_at"] = moment.isoformat() + if failures: + payload["rejection_codes"] = sorted(failures) + if current_receipt is not None: + leaf = ReportingAdjustmentReceipt.model_validate( + current_receipt.model_dump(mode="json", exclude_none=True, warnings="error") + ) + wire = leaf.model_dump(mode="json", exclude_none=True) + if ( + not _schema_valid(wire, "reporting-adjustment-receipt") + or leaf.reporting_adjustment_id != adjustment.reporting_adjustment_id + or leaf.adjusts_reporting_revision_id != adjustment.adjusts_reporting_revision_id + or leaf.reporting_receipt_id == reporting_receipt_id + or leaf.observed_at > moment + ): + _fail("INVALID_RECEIPT") + if str(leaf.status) == "accepted": + _fail("RECEIPT_TERMINAL") + payload["supersedes_reporting_receipt_id"] = leaf.reporting_receipt_id + if _schema_valid(payload, "reporting-adjustment-receipt"): + receipt = ReportingAdjustmentReceipt.model_validate(payload) + except ReportingAdjustmentEvidenceError: + raise + except (ValueError, TypeError, AttributeError, OverflowError): + # Raise the closed receipt error outside the exception handler. + receipt = None + if receipt is None: + _fail("INVALID_RECEIPT") + return receipt diff --git a/tests/test_reporting_adjustment_evidence.py b/tests/test_reporting_adjustment_evidence.py new file mode 100644 index 000000000..81befbf9c --- /dev/null +++ b/tests/test_reporting_adjustment_evidence.py @@ -0,0 +1,642 @@ +"""Pure evidence/receipt contracts, independent of transport and history selection.""" + +from __future__ import annotations + +import hashlib +import json +import traceback +from copy import deepcopy +from dataclasses import FrozenInstanceError, replace +from datetime import datetime, timezone +from typing import Any + +import pytest +import rfc8785 + +from adcp.reporting.adjustment_evidence import ( + ReportingAdjustmentEvidence, + ReportingAdjustmentEvidenceError, + ReportingAdjustmentEvidenceLimits, + ReportingAdjustmentReceiptContext, + ReportingAdjustmentScope, + build_reporting_adjustment_receipt, + capture_reporting_adjustment_evidence, +) +from adcp.types import ( + ReportingAdjustment, + ReportingAdjustmentReceipt, + ReportingObligation, + ReportingRevision, +) +from adcp.validation.schema_loader import get_named_validator + +SECRET = "private-body-marker" +SCOPE = ReportingAdjustmentScope( + "https://seller.example/agent", "account-1", "consumer-1", "obligation-1" +) +NOW = datetime(2026, 9, 3, tzinfo=timezone.utc) +CONTEXT = ReportingAdjustmentReceiptContext( + SCOPE, "revision-official", datetime(2026, 9, 2, tzinfo=timezone.utc), (("spend", "USD"),) +) + + +def adjustment(**changes: Any) -> dict[str, Any]: + raw = { + "reporting_adjustment_id": "adjustment-1", + "adjusts_reporting_revision_id": "revision-official", + "reason_code": "source_correction", + "accounting_period": {"start": "2026-09-01T00:00:00Z", "end": "2026-10-01T00:00:00Z"}, + "control_total_deltas": [ + {"name": "spend", "value": "-0.00", "value_type": "decimal", "unit": "USD"} + ], + "correction_observed_at": "2026-09-02T05:00:00.000+05:00", + "created_at": "2026-09-02T00:00:00.000000Z", + } + raw.update(changes) + raw["canonical_adjustment_sha256"] = hashlib.sha256(rfc8785.dumps(raw)).hexdigest() + return raw + + +def capture(raw: dict[str, Any] | None = None, **kwargs: Any) -> ReportingAdjustmentEvidence: + raw = adjustment() if raw is None else raw + return capture_reporting_adjustment_evidence( + raw, typed_adjustment=ReportingAdjustment.model_validate(raw), scope=SCOPE, **kwargs + ) + + +def build( + evidence: ReportingAdjustmentEvidence | None = None, **kwargs: Any +) -> ReportingAdjustmentReceipt: + return build_reporting_adjustment_receipt( + capture() if evidence is None else evidence, + kwargs.pop("context", CONTEXT), + reporting_receipt_id=kwargs.pop("reporting_receipt_id", "receipt-adjustment-0001"), + observed_at=kwargs.pop("observed_at", NOW), + **kwargs, + ) + + +def rejected(code: str, call: Any) -> None: + with pytest.raises(ReportingAdjustmentEvidenceError) as caught: + call() + assert caught.value.code == code + assert caught.value.args == (code,) + assert SECRET not in repr(caught.value) + assert SECRET not in "".join(traceback.format_exception(caught.type, caught.value, caught.tb)) + + +@pytest.mark.parametrize("detail", [None, "café", "cafe\u0301", "accounting \U0001f4b0"]) +def test_raw_values_and_presence_are_hashed_before_model_normalization(detail: str | None) -> None: + raw = adjustment(**({} if detail is None else {"reason_detail": detail})) + typed = ReportingAdjustment.model_validate(raw) + wire = json.dumps(raw, ensure_ascii=False, indent=2).encode() + evidence = capture_reporting_adjustment_evidence(wire, typed_adjustment=typed, scope=SCOPE) + expected = dict(raw) + expected.pop("canonical_adjustment_sha256") + assert evidence.raw_json == wire + assert evidence.canonical_json == rfc8785.dumps(expected) + assert evidence.observed_adjustment_sha256 == raw["canonical_adjustment_sha256"] + assert evidence.input_kind == "bytes" + assert evidence.adjustment == typed + normalized = typed.model_dump(mode="json", exclude_none=True) + normalized.pop("canonical_adjustment_sha256") + assert rfc8785.dumps(normalized) != evidence.canonical_json + assert json.loads(evidence.canonical_json)["control_total_deltas"][0]["value"] == "-0.00" + assert json.loads(evidence.canonical_json)["control_total_deltas"][0]["value_type"] == "decimal" + + +def test_key_order_is_irrelevant_but_optional_presence_unicode_and_array_order_are_not() -> None: + raw = adjustment() + assert ( + capture(dict(reversed(list(raw.items())))).observed_adjustment_sha256 + == capture(raw).observed_adjustment_sha256 + ) + without_unit = deepcopy(raw) + del without_unit["control_total_deltas"][0]["unit"] + assert ( + capture(without_unit).observed_adjustment_sha256 != capture(raw).observed_adjustment_sha256 + ) + assert ( + capture(adjustment(reason_detail="é")).observed_adjustment_sha256 + != capture(adjustment(reason_detail="e\u0301")).observed_adjustment_sha256 + ) + two = adjustment( + control_total_deltas=[ + {"name": "spend", "value": "1", "value_type": "decimal", "unit": "USD"}, + {"name": "fee", "value": "2", "value_type": "integer"}, + ] + ) + reversed_deltas = deepcopy(two) + reversed_deltas["control_total_deltas"].reverse() + assert ( + capture(two).observed_adjustment_sha256 + != capture(reversed_deltas).observed_adjustment_sha256 + ) + + +def test_mapping_is_snapshot_with_explicit_upstream_information_loss_limit() -> None: + raw = adjustment(reason_detail=SECRET) + typed = ReportingAdjustment.model_validate(raw) + evidence = capture_reporting_adjustment_evidence(raw, typed_adjustment=typed, scope=SCOPE) + before = evidence.raw_json, evidence.canonical_json, evidence.observed_adjustment_sha256 + raw["control_total_deltas"][0]["value"] = "999" + typed.control_total_deltas[0].root.value = "999" + evidence.adjustment.control_total_deltas[0].root.value = "999" + assert ( + evidence.raw_json, + evidence.canonical_json, + evidence.observed_adjustment_sha256, + ) == before + assert evidence.input_kind == "mapping" + assert evidence.adjustment.control_total_deltas[0].value == "-0.00" + assert SECRET not in repr(evidence) + assert "seller.example" not in repr(evidence.scope) + with pytest.raises(FrozenInstanceError): + evidence.raw_json = b"{}" + restored = ReportingAdjustmentEvidence(evidence.scope, evidence.raw_json, evidence.input_kind) + assert restored == evidence + + +@pytest.mark.parametrize( + "wire", + [ + b'{"reason_detail":"private-body-marker","reason_detail":"second"}', + b'{"nested":{"value":"private-body-marker","value":"second"}}', + b'{"reason_detail":"private-body-marker","bad":NaN}', + b'{"reason_detail":"private-body-marker","bad":Infinity}', + b'{"reason_detail":"private-body-marker","bad":9007199254740992}', + b'{"reason_detail":"private-body-marker","bad":1e400}', + b'{"reason_detail":"private-body-marker","bad":"\\ud800"}', + b'{"reason_detail":"private-body-marker","bad":"\xff"}', + b'"private-body-marker"', + b"[]", + b"null", + b'{"private-body-marker":', + ], +) +def test_strict_byte_admission_rejects_ambiguous_or_malformed_json(wire: bytes) -> None: + rejected("INVALID_EVIDENCE", lambda: ReportingAdjustmentEvidence(SCOPE, wire, "bytes")) + + +def test_decoded_mapping_cannot_recover_duplicate_keys() -> None: + raw = adjustment(reason_detail="safe") + wire = ( + json.dumps(raw) + .replace('"reason_detail": "safe"', '"reason_detail": "discarded", "reason_detail": "safe"') + .encode() + ) + rejected("INVALID_EVIDENCE", lambda: ReportingAdjustmentEvidence(SCOPE, wire, "bytes")) + evidence = capture(json.loads(wire)) + assert evidence.input_kind == "mapping" + assert evidence.adjustment.reason_detail == "safe" + + +@pytest.mark.parametrize( + "changes", + [ + {"reason_detail": None}, + {"extra": SECRET}, + {"reason_code": SECRET}, + {"control_total_deltas": [{"name": "spend", "value": 1}]}, + { + "control_total_deltas": [ + {"name": "spend", "value": "1"}, + {"name": "spend", "value": "2"}, + ] + }, + {"accounting_period": {"start": "2026-10-01T00:00:00Z", "end": "2026-09-01T00:00:00Z"}}, + {"created_at": "2026-09-01T00:00:00Z"}, + {"created_at": "2026-09-02T00:00:00"}, + ], +) +def test_schema_and_cross_field_admission(changes: dict[str, Any]) -> None: + raw = adjustment(**changes) + rejected( + "INVALID_EVIDENCE", + lambda: ReportingAdjustmentEvidence(SCOPE, json.dumps(raw).encode(), "bytes"), + ) + + +def test_missing_advertised_digest_cannot_be_accepted_as_reconciled_evidence() -> None: + raw = adjustment() + del raw["canonical_adjustment_sha256"] + rejected("INVALID_EVIDENCE", lambda: capture(raw)) + + +def test_temporal_checks_preserve_sub_microsecond_precision() -> None: + raw = adjustment( + correction_observed_at="2026-09-02T00:00:00.0000002Z", + created_at="2026-09-02T00:00:00.0000001Z", + ) + # Both become the same datetime in the model; raw evidence still rejects their order. + typed = ReportingAdjustment.model_validate(raw) + assert typed.correction_observed_at == typed.created_at + rejected("INVALID_EVIDENCE", lambda: capture(raw)) + raw = adjustment( + accounting_period={ + "start": "2026-09-01T00:00:00.0000001Z", + "end": "2026-09-01T00:00:00.0000002Z", + }, + created_at="2026-09-03T00:00:00.0000001Z", + ) + evidence = capture(raw) + rejected("INVALID_RECEIPT", lambda: build(evidence)) + assert build(evidence, observed_at=NOW.replace(microsecond=1)).status == "accepted" + + +def test_generated_default_does_not_repair_a_missing_required_wire_field() -> None: + raw = adjustment() + del raw["control_total_deltas"][0]["value_type"] + # Generated models accept a default here; the authoritative wire schema does not. + ReportingAdjustment.model_validate(raw) + rejected("INVALID_EVIDENCE", lambda: capture(raw)) + + +@pytest.mark.parametrize( + "limits", + [ + ReportingAdjustmentEvidenceLimits(max_bytes=100), + ReportingAdjustmentEvidenceLimits(max_depth=2), + ReportingAdjustmentEvidenceLimits(max_nodes=10), + ], +) +def test_byte_and_mapping_admission_share_explicit_resource_bounds( + limits: ReportingAdjustmentEvidenceLimits, +) -> None: + raw = adjustment() + rejected("EVIDENCE_LIMIT_EXCEEDED", lambda: capture(raw, limits=limits)) + rejected( + "EVIDENCE_LIMIT_EXCEEDED", + lambda: ReportingAdjustmentEvidence(SCOPE, json.dumps(raw).encode(), "bytes", limits), + ) + + +def test_cycles_invalid_runtime_types_and_bad_limits_fail_closed() -> None: + raw = adjustment() + raw["loop"] = raw + rejected( + "INVALID_EVIDENCE", + lambda: capture_reporting_adjustment_evidence( + raw, typed_adjustment=ReportingAdjustment.model_validate(adjustment()), scope=SCOPE + ), + ) + rejected("INVALID_EVIDENCE", lambda: capture(limits=None)) + rejected("INVALID_EVIDENCE", lambda: ReportingAdjustmentEvidenceLimits(max_depth=0)) + raw = adjustment() + raw["reason_detail"] = object() + rejected( + "INVALID_EVIDENCE", + lambda: capture_reporting_adjustment_evidence( + raw, typed_adjustment=ReportingAdjustment.model_validate(adjustment()), scope=SCOPE + ), + ) + + +@pytest.mark.parametrize( + "field,value", + [("reporting_adjustment_id", "another"), ("reason_detail", SECRET), ("created_at", "bad")], +) +def test_mutated_or_unvalidated_typed_view_cannot_disagree_with_raw(field: str, value: str) -> None: + raw = adjustment() + typed = ReportingAdjustment.model_validate(raw).model_copy(update={field: value}) + rejected( + "TYPED_EVIDENCE_MISMATCH", + lambda: capture_reporting_adjustment_evidence(raw, typed_adjustment=typed, scope=SCOPE), + ) + + +def test_schema_unavailability_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("adcp.reporting.adjustment_evidence.get_named_validator", lambda _: None) + rejected("SCHEMA_UNAVAILABLE", capture) + + +def test_resolver_failure_does_not_leak_evidence(monkeypatch: pytest.MonkeyPatch) -> None: + def broken(_: str) -> None: + raise RuntimeError(SECRET) + + monkeypatch.setattr("adcp.reporting.adjustment_evidence.get_named_validator", broken) + rejected("SCHEMA_UNAVAILABLE", capture) + + +def test_receipt_identity_time_digest_and_wire_shape_are_deterministic() -> None: + receipt = build() + assert receipt == build() + wire = receipt.model_dump(mode="json", exclude_none=True) + assert wire == { + "reporting_receipt_id": "receipt-adjustment-0001", + "reporting_adjustment_id": "adjustment-1", + "adjusts_reporting_revision_id": "revision-official", + "status": "accepted", + "observed_adjustment_sha256": capture().observed_adjustment_sha256, + "observed_at": "2026-09-03T00:00:00Z", + } + validator = get_named_validator("core/reporting-adjustment-receipt.json") + assert validator is not None and validator.is_valid(wire) + assert not validator.is_valid({**wire, "rejection_codes": ["ADJUSTMENT_DIGEST_MISMATCH"]}) + assert not validator.is_valid({**wire, "status": "rejected"}) + + +def test_digest_mismatch_and_caller_semantic_rejection_use_closed_unique_codes() -> None: + raw = adjustment() + raw["canonical_adjustment_sha256"] = "0" * 64 + receipt = build( + capture(raw), + rejection_codes=("ADJUSTMENT_SEMANTIC_MISMATCH", "ADJUSTMENT_SEMANTIC_MISMATCH"), + ) + assert str(receipt.status) == "rejected" + assert receipt.model_dump(mode="json")["rejection_codes"] == [ + "ADJUSTMENT_DIGEST_MISMATCH", + "ADJUSTMENT_SEMANTIC_MISMATCH", + ] + assert receipt.observed_adjustment_sha256 == capture().observed_adjustment_sha256 + rejected("INVALID_RECEIPT", lambda: build(rejection_codes=(SECRET,))) + + +@pytest.mark.parametrize( + "field,value", + [ + ("seller_identity", "https://other.example/agent"), + ("account_id", "other"), + ("consumer_id", "other"), + ("reporting_obligation_id", "other"), + ], +) +def test_cross_scope_context_never_reuses_evidence(field: str, value: str) -> None: + other = replace(CONTEXT, scope=replace(SCOPE, **{field: value})) + rejected("ADJUSTMENT_CONTEXT_MISMATCH", lambda: build(context=other)) + + +@pytest.mark.parametrize( + "context", + [ + replace(CONTEXT, reporting_revision_id="wrong"), + replace(CONTEXT, finalized_at=NOW), + replace(CONTEXT, control_total_units=(("spend", "EUR"),)), + replace(CONTEXT, control_total_units=(("another", "USD"),)), + ], +) +def test_exact_target_official_lock_and_delta_units_are_required( + context: ReportingAdjustmentReceiptContext, +) -> None: + rejected("ADJUSTMENT_CONTEXT_MISMATCH", lambda: build(context=context)) + + +def test_maximum_consumer_identity_is_preserved_without_normalization() -> None: + prefix = "https://consumer.example/" + consumer = prefix + "x" * (2048 - len(prefix)) + scope = replace(SCOPE, consumer_id=consumer) + evidence = capture_reporting_adjustment_evidence( + adjustment(), typed_adjustment=ReportingAdjustment.model_validate(adjustment()), scope=scope + ) + assert build(evidence, context=replace(CONTEXT, scope=scope)).status == "accepted" + assert evidence.scope.consumer_id == consumer + rejected("INVALID_CONTEXT", lambda: replace(scope, consumer_id=consumer + "x")) + + +@pytest.mark.parametrize( + "kwargs", + [ + {"observed_at": datetime(2026, 9, 3)}, + {"observed_at": datetime(2026, 9, 1, tzinfo=timezone.utc)}, + {"reporting_receipt_id": "short"}, + {"reporting_receipt_id": SECRET + " bad"}, + {"reporting_receipt_id": SECRET + "\n"}, + ], +) +def test_bad_receipt_identity_or_time_never_escapes_as_model_error(kwargs: dict[str, Any]) -> None: + rejected("INVALID_RECEIPT", lambda: build(**kwargs)) + + +def test_current_rejected_leaf_is_superseded_exactly_and_accepted_is_terminal() -> None: + prior = build(rejection_codes=("ADJUSTMENT_SEMANTIC_MISMATCH",)) + receipt = build(reporting_receipt_id="receipt-adjustment-0002", current_receipt=prior) + assert receipt.supersedes_reporting_receipt_id == prior.reporting_receipt_id + assert receipt.status == "accepted" + rejected( + "RECEIPT_TERMINAL", + lambda: build(reporting_receipt_id="receipt-adjustment-0003", current_receipt=receipt), + ) + + +@pytest.mark.parametrize( + "updates", + [ + {"reporting_adjustment_id": "another"}, + {"adjusts_reporting_revision_id": "another"}, + {"rejection_codes": None}, + {"observed_at": datetime(2026, 9, 4, tzinfo=timezone.utc)}, + ], +) +def test_invalid_or_foreign_current_leaf_is_rejected(updates: dict[str, Any]) -> None: + prior = build(rejection_codes=("ADJUSTMENT_SEMANTIC_MISMATCH",)).model_copy(update=updates) + rejected( + "INVALID_RECEIPT", + lambda: build(reporting_receipt_id="receipt-adjustment-0002", current_receipt=prior), + ) + + +def test_receipt_id_cannot_supersede_itself() -> None: + rejected( + "INVALID_RECEIPT", + lambda: build(current_receipt=build(rejection_codes=("ADJUSTMENT_SEMANTIC_MISMATCH",))), + ) + + +def selected() -> tuple[ReportingObligation, ReportingRevision]: + # Public model fixtures remain independent of transport and ledger test helpers. + period = { + "start": "2026-08-01T00:00:00Z", + "end": "2026-09-01T00:00:00Z", + "source_timezone": "UTC", + } + coverage = { + "status": "full", + "evaluated_at": period["end"], + "media_buy_ids": ["buy-1"], + "fully_covered_media_buy_ids": ["buy-1"], + "partially_covered_media_buy_ids": [], + "unsupported_media_buy_ids": [], + "unknown_media_buy_ids": [], + "package_ids": [], + "covered_package_ids": [], + "unsupported_package_ids": [], + "unknown_package_ids": [], + "limitations": [], + } + shared = { + "report_definition_id": "billing-v1", + "reporting_profile": "billing-v1", + "account_id": "account-1", + "media_buy_ids": ["buy-1"], + "coverage": coverage, + "period": period, + } + obligation = ReportingObligation.model_validate( + { + **shared, + "reporting_obligation_id": "obligation-1", + "delivery_config_id": "billing-feed", + "delivery_config_version": 1, + "feed_purpose": "billing", + "scope_resolved_at": period["end"], + "expected_at": "2026-09-02T00:00:00Z", + "schedule": { + "period_duration": "P1M", + "alignment": "billing_cycle", + "delivery_sla": "P1D", + }, + "destination_ref": "destination-1", + "required_finality": "official", + "reconciliation_mode": "consumer_receipt", + "reconciliation_status": "pending", + "health": "waiting", + "production_status": "published", + "revision_count": 1, + "materialization_count": 1, + "successful_materialization_count": 1, + "receipt_count": 0, + "accepted_receipt_count": 0, + "issues": [], + "resource_retained_until": "2026-12-01T00:00:00Z", + } + ) + revision = ReportingRevision.model_validate( + { + **shared, + "reporting_revision_id": "revision-official", + "revision_content_sha256": "e" * 64, + "report_definition_uri": "https://schemas.example/billing.json", + "report_definition_sha256": "d" * 64, + "schema_version": "1", + "schema_uri": "https://schemas.example/billing.json", + "schema_sha256": "c" * 64, + "schema_dialect": "https://json-schema.org/draft/2020-12/schema", + "schema_ref_policy": "local_fragment_only", + "finality": "official", + "finality_basis": "source_final", + "finality_policy_id": "source-final", + "finalized_at": "2026-09-02T00:00:00Z", + "observed_at": "2026-09-02T00:00:00Z", + "data_through": period["end"], + "data_through_precision": "exact", + "row_count": 7, + "control_totals": [ + {"name": "spend", "value": "7000.00", "value_type": "decimal", "unit": "USD"} + ], + "canonical_content_digest": { + "algorithm": "sha256", + "value": "a" * 64, + "canonicalization_id": "rows-v1", + "canonicalization_uri": "https://schemas.example/rows.json", + "canonicalization_sha256": "b" * 64, + }, + "created_at": "2026-09-02T00:00:00Z", + } + ) + return obligation, revision + + +def test_selected_context_requires_exact_ownership_and_reconciled_billing_official() -> None: + obligation, revision = selected() + context = ReportingAdjustmentReceiptContext.from_selection( + SCOPE, obligation=obligation, revision=revision, revision_owner="obligation-1" + ) + assert context == CONTEXT + revision.control_totals[0].root.unit = "EUR" + assert context.control_total_units == (("spend", "USD"),) + assert build(context=context).status == "accepted" + rejected( + "INVALID_CONTEXT", + lambda: ReportingAdjustmentReceiptContext.from_selection( + SCOPE, obligation=obligation, revision=revision, revision_owner="other" + ), + ) + + +@pytest.mark.parametrize( + "kind,updates", + [ + ("obligation", {"account_id": "other"}), + ("revision", {"account_id": "other"}), + ("obligation", {"feed_purpose": "performance"}), + ("obligation", {"reconciliation_mode": "none"}), + ("revision", {"finality": "provisional"}), + ("revision", {"finalized_at": None}), + ("revision", {"report_definition_id": "other"}), + ("revision", {"reporting_profile": "other"}), + ( + "revision", + { + "period": { + "start": "2026-07-01T00:00:00Z", + "end": "2026-08-01T00:00:00Z", + "source_timezone": "UTC", + } + }, + ), + ], +) +def test_selection_binding_cannot_be_inferred_from_an_unrelated_official( + kind: str, updates: dict[str, Any] +) -> None: + obligation, revision = selected() + if kind == "obligation": + obligation = obligation.model_copy(update=updates) + else: + revision = revision.model_copy(update=updates) + rejected( + "INVALID_CONTEXT", + lambda: ReportingAdjustmentReceiptContext.from_selection( + SCOPE, obligation=obligation, revision=revision, revision_owner="obligation-1" + ), + ) + + +@pytest.mark.parametrize("side", ["obligation", "revision"]) +def test_selected_period_requires_exact_source_timezone(side: str) -> None: + obligation, revision = selected() + target = obligation if side == "obligation" else revision + target.period.source_timezone = "America/New_York" + rejected( + "INVALID_CONTEXT", + lambda: ReportingAdjustmentReceiptContext.from_selection( + SCOPE, obligation=obligation, revision=revision, revision_owner="obligation-1" + ), + ) + + +@pytest.mark.parametrize("units", [("ab",), ("spendUSD",), ({"spend": "USD"},)]) +def test_context_unit_pairs_do_not_coerce_strings_or_mappings(units: Any) -> None: + rejected("INVALID_CONTEXT", lambda: replace(CONTEXT, control_total_units=units)) + + +def test_context_detaches_explicit_list_pairs() -> None: + units = [["spend", "USD"]] + context = replace(CONTEXT, control_total_units=units) + units[0][1] = "EUR" + assert context.control_total_units == (("spend", "USD"),) + + +def test_receipt_model_validation_failure_is_closed(monkeypatch: pytest.MonkeyPatch) -> None: + evidence = capture() + + def fail(*args: Any, **kwargs: Any) -> None: + raise ValueError(SECRET) + + monkeypatch.setattr(ReportingAdjustmentReceipt, "model_validate", fail) + with pytest.raises(ReportingAdjustmentEvidenceError) as caught: + build(evidence) + assert caught.value.code == "INVALID_RECEIPT" + assert caught.value.__context__ is None and caught.value.__cause__ is None + assert SECRET not in "".join(traceback.format_exception(caught.type, caught.value, caught.tb)) + + +def test_uppercase_advertised_digest_preserves_raw_evidence_and_compares_hex_value() -> None: + raw = adjustment() + expected = raw["canonical_adjustment_sha256"] + raw["canonical_adjustment_sha256"] = expected.upper() + evidence = capture(raw) + receipt = build(evidence) + assert json.loads(evidence.raw_json)["canonical_adjustment_sha256"] == expected.upper() + assert receipt.observed_adjustment_sha256 == expected + assert receipt.status == "accepted" diff --git a/tests/type_checks/reporting_adjustment_evidence.py b/tests/type_checks/reporting_adjustment_evidence.py new file mode 100644 index 000000000..18ff6b509 --- /dev/null +++ b/tests/type_checks/reporting_adjustment_evidence.py @@ -0,0 +1,47 @@ +"""Standalone public imports for the pure buyer adjustment primitive.""" + +from datetime import datetime + +from adcp.reporting.adjustment_evidence import ( + ReportingAdjustmentEvidence, + ReportingAdjustmentReceiptContext, + ReportingAdjustmentScope, + build_reporting_adjustment_receipt, + capture_reporting_adjustment_evidence, +) +from adcp.types import ( + ReportingAdjustment, + ReportingAdjustmentReceipt, + ReportingObligation, + ReportingRevision, +) + + +def capture_and_build( + raw: bytes, + typed: ReportingAdjustment, + trusted_scope: ReportingAdjustmentScope, + selected_obligation: ReportingObligation, + selected_official: ReportingRevision, + exact_revision_owner: str, + reserved_receipt_id: str, + reserved_observed_at: datetime, + verified_current_leaf: ReportingAdjustmentReceipt | None, +) -> tuple[ReportingAdjustmentEvidence, ReportingAdjustmentReceipt]: + evidence = capture_reporting_adjustment_evidence( + raw, typed_adjustment=typed, scope=trusted_scope + ) + context = ReportingAdjustmentReceiptContext.from_selection( + trusted_scope, + obligation=selected_obligation, + revision=selected_official, + revision_owner=exact_revision_owner, + ) + receipt = build_reporting_adjustment_receipt( + evidence, + context, + reporting_receipt_id=reserved_receipt_id, + observed_at=reserved_observed_at, + current_receipt=verified_current_leaf, + ) + return evidence, receipt