diff --git a/src/govoplan_payments/backend/dsar_provider.py b/src/govoplan_payments/backend/dsar_provider.py new file mode 100644 index 0000000..3987eff --- /dev/null +++ b/src/govoplan_payments/backend/dsar_provider.py @@ -0,0 +1,550 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_payments.backend.db.models import ( + PaymentEvent, + PaymentObligation, + PaymentReconciliation, +) + + +PAYMENTS_DSAR_CAPABILITY = dsar_capability_name("payments") +_MAX_RECORDS = 5_000 +_MAX_CHILD_RECORDS = 1_000 +_CONFLICT = object() +_ATTRIBUTION_TYPES = frozenset( + { + "payment_request_attribution", + "payment_reconciliation_attribution", + "payment_event_attribution", + } +) + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + actor_refs: tuple[str, ...] + payment_row_id: str | None + payment_id: str | None + payment_reference: str | None + + @property + def has_payment_selector(self) -> bool: + return bool(self.payment_row_id or self.payment_id or self.payment_reference) + + +class PaymentsDsarProvider: + provider_id = "payments" + module_id = "payments" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + return () + + if selectors.has_payment_selector: + query = db.query(PaymentObligation).filter( + PaymentObligation.tenant_id == tenant_id + ) + if selectors.payment_row_id: + query = query.filter(PaymentObligation.id == selectors.payment_row_id) + if selectors.payment_id: + query = query.filter( + PaymentObligation.payment_id == selectors.payment_id + ) + if selectors.payment_reference: + query = query.filter( + PaymentObligation.payment_reference == selectors.payment_reference + ) + rows = ( + query.order_by( + PaymentObligation.requested_at, + PaymentObligation.id, + ) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Payments DSAR result limit exceeded; narrow the identifiers." + ) + return tuple(_payment_record(db, row) for row in rows) + + if not selectors.actor_refs: + return () + records: list[DsarRecordRef] = [] + obligations = db.query(PaymentObligation).filter( + PaymentObligation.tenant_id == tenant_id, + PaymentObligation.requested_by_ref.in_(selectors.actor_refs), + ) + records.extend( + _request_attribution(row) + for row in _limited( + obligations, + PaymentObligation, + "request attribution", + ) + ) + reconciliations = db.query(PaymentReconciliation).filter( + PaymentReconciliation.tenant_id == tenant_id, + PaymentReconciliation.recorded_by_ref.in_(selectors.actor_refs), + ) + records.extend( + _reconciliation_attribution(row) + for row in _limited( + reconciliations, + PaymentReconciliation, + "reconciliation attribution", + ) + ) + events = db.query(PaymentEvent).filter( + PaymentEvent.tenant_id == tenant_id, + PaymentEvent.actor_ref.in_(selectors.actor_refs), + ) + records.extend( + _event_attribution(row) + for row in _limited(events, PaymentEvent, "event attribution") + ) + if len(records) > _MAX_RECORDS: + raise ValueError( + "Payments DSAR combined result limit exceeded; narrow the selectors." + ) + order = { + "payment_request_attribution": 10, + "payment_reconciliation_attribution": 20, + "payment_event_attribution": 30, + } + return tuple( + sorted( + records, + key=lambda item: (order[item.resource_type], item.resource_id), + ) + ) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del tenant_id + _session(session) + if _subject_selectors(subject) is None: + raise ValueError("Payments DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + actions.append( + DsarErasureActionRef( + action_id=( + f"payments:retain:{record.resource_type}:{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind="retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Retain {record.title}", + rationale=record.retention_reason + or "Financial and reconciliation evidence must be retained.", + executable=False, + ) + ) + return tuple(actions) + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: + del tenant_id + _session(session) + if _subject_selectors(subject) is None: + raise ValueError("Payments DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "retain": + raise ValueError("Payments DSAR publishes retain-only actions.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Payment and reconciliation evidence remains under the " + "configured financial, statutory, and legal-hold policy." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + references = subject.external_references + values = { + "account_id": _coalesce( + subject.account_id, + references.get("payments.account"), + references.get("access.account"), + ), + "membership_id": _coalesce( + subject.membership_id, + references.get("payments.membership"), + references.get("tenancy.membership"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("payments.identity"), + references.get("identity.id"), + ), + "actor_ref": _coalesce( + references.get("payments.actor"), + references.get("payments.operator"), + ), + "payment_row_id": _coalesce( + references.get("payments.obligation"), + references.get("payments.row"), + ), + "payment_id": _coalesce( + references.get("payments.payment"), + references.get("payments.payment_id"), + ), + "payment_reference": _coalesce( + references.get("payments.reference"), + references.get("payments.payment_reference"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + account_id = _optional_string(values["account_id"]) + membership_id = _optional_string(values["membership_id"]) + identity_id = _optional_string(values["identity_id"]) + direct_actor = _optional_string(values["actor_ref"]) + actor_refs = tuple( + dict.fromkeys( + value + for value in ( + account_id, + f"account:{account_id}" if account_id else None, + membership_id, + f"membership:{membership_id}" if membership_id else None, + identity_id, + f"identity:{identity_id}" if identity_id else None, + direct_actor, + ) + if value + ) + ) + selectors = _SubjectSelectors( + actor_refs=actor_refs, + payment_row_id=_optional_string(values["payment_row_id"]), + payment_id=_optional_string(values["payment_id"]), + payment_reference=_optional_string(values["payment_reference"]), + ) + if not selectors.actor_refs and not selectors.has_payment_selector: + return None + return selectors + + +def _coalesce(*values: str | None) -> str | None | object: + normalized = {str(value).strip() for value in values if str(value or "").strip()} + if len(normalized) > 1: + return _CONFLICT + return next(iter(normalized), None) + + +def _optional_string(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _limited(query, model, label: str): + rows = query.order_by(model.created_at, model.id).limit(_MAX_RECORDS + 1).all() + if len(rows) > _MAX_RECORDS: + raise ValueError(f"Payments DSAR {label} limit exceeded; narrow the selectors.") + return rows + + +def _payment_record( + session: Session, + obligation: PaymentObligation, +) -> DsarRecordRef: + reconciliations = _children( + session.query(PaymentReconciliation).filter( + PaymentReconciliation.tenant_id == obligation.tenant_id, + PaymentReconciliation.payment_row_id == obligation.id, + ), + PaymentReconciliation, + "reconciliation", + ) + events = _children( + session.query(PaymentEvent).filter( + PaymentEvent.tenant_id == obligation.tenant_id, + PaymentEvent.payment_row_id == obligation.id, + ), + PaymentEvent, + "event", + ) + return DsarRecordRef( + provider_id="payments", + module_id="payments", + resource_type="payment_obligation", + resource_id=obligation.id, + category="financial_obligation_and_evidence", + title=f"Payment obligation {obligation.payment_reference}", + data={ + "payment_id": obligation.payment_id, + "payment_reference": obligation.payment_reference, + "source_module": obligation.source_module, + "source_resource_type": obligation.source_resource_type, + "source_resource_id": obligation.source_resource_id, + "amount_minor": obligation.amount_minor, + "currency": obligation.currency, + "subject": obligation.subject[:1_000], + "status": obligation.status, + "requested_at": _iso(obligation.requested_at), + "requested_by_ref": obligation.requested_by_ref, + "due_at": _iso(obligation.due_at), + "settled_at": _iso(obligation.settled_at), + "context_refs": _context_refs(obligation.context_refs), + "reconciliations": [ + { + "id": row.id, + "reconciliation_id": row.reconciliation_id, + "mode": row.mode, + "amount_minor": row.amount_minor, + "currency": row.currency, + "transaction_reference": row.transaction_reference, + "evidence_ref": _evidence_reference(row.evidence_ref), + "received_at": _iso(row.received_at), + "recorded_at": _iso(row.recorded_at), + "recorded_by_ref": row.recorded_by_ref, + } + for row in reconciliations + ], + "events": [ + { + "id": row.id, + "event_id": row.event_id, + "event_type": row.event_type, + "status": row.status, + "occurred_at": _iso(row.occurred_at), + "actor_ref": row.actor_ref, + } + for row in events + ], + }, + observed_at=_aware(obligation.updated_at), + immutable_evidence=True, + retention_reason=( + "The exact obligation, reconciliation, and lifecycle records are " + "financial evidence. Arbitrary metadata and event payloads are excluded." + ), + ) + + +def _children(query, model, label: str): + rows = ( + query.order_by(model.created_at, model.id).limit(_MAX_CHILD_RECORDS + 1).all() + ) + if len(rows) > _MAX_CHILD_RECORDS: + raise ValueError( + f"Payment {label} history exceeds the DSAR bound; narrow and review the payment." + ) + return rows + + +def _request_attribution(row: PaymentObligation) -> DsarRecordRef: + return _attribution_record( + "payment_request_attribution", + row.id, + "Requested payment obligation", + { + "activity": "requested_payment_obligation", + "payment_id": row.payment_id, + "payment_reference": row.payment_reference, + "source_module": row.source_module, + "source_resource_type": row.source_resource_type, + "source_resource_id": row.source_resource_id, + "amount_minor": row.amount_minor, + "currency": row.currency, + "status": row.status, + "requested_at": _iso(row.requested_at), + }, + row.requested_at, + ) + + +def _reconciliation_attribution(row: PaymentReconciliation) -> DsarRecordRef: + return _attribution_record( + "payment_reconciliation_attribution", + row.id, + "Recorded payment reconciliation", + { + "activity": "recorded_payment_reconciliation", + "payment_row_id": row.payment_row_id, + "reconciliation_id": row.reconciliation_id, + "mode": row.mode, + "amount_minor": row.amount_minor, + "currency": row.currency, + "received_at": _iso(row.received_at), + "recorded_at": _iso(row.recorded_at), + }, + row.recorded_at, + ) + + +def _event_attribution(row: PaymentEvent) -> DsarRecordRef: + return _attribution_record( + "payment_event_attribution", + row.id, + "Payment lifecycle event attribution", + { + "activity": "recorded_payment_event", + "payment_row_id": row.payment_row_id, + "event_id": row.event_id, + "event_type": row.event_type, + "status": row.status, + "occurred_at": _iso(row.occurred_at), + }, + row.occurred_at, + ) + + +def _attribution_record( + resource_type: str, + resource_id: str, + title: str, + data: Mapping[str, object], + observed_at: datetime, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="payments", + module_id="payments", + resource_type=resource_type, + resource_id=resource_id, + category="operator_accountability_evidence", + title=title, + data=data, + observed_at=_aware(observed_at), + immutable_evidence=True, + retention_reason=( + "Payment operator attribution is financial accountability evidence; " + "arbitrary metadata, hashes, replay keys, and payloads are excluded." + ), + ) + + +def _context_refs(value: object) -> dict[str, str]: + if not isinstance(value, Mapping) or len(value) > 100: + raise ValueError("Payment context references exceed the DSAR bound.") + result: dict[str, str] = {} + for raw_key, raw_value in value.items(): + key = str(raw_key) + if not key or len(key) > 200: + raise ValueError("Payment context reference key is invalid.") + result[key] = "[redacted]" if _sensitive_key(key) else str(raw_value)[:2_000] + return result + + +def _evidence_reference(value: object) -> dict[str, object]: + if not isinstance(value, Mapping): + raise ValueError("Payment evidence reference is invalid.") + derived = value.get("derived_from") + if not isinstance(derived, list) or len(derived) > 100: + raise ValueError("Payment evidence derivation exceeds the DSAR bound.") + return { + "kind": _bounded(value.get("kind"), 100), + "owner_module": _bounded(value.get("owner_module"), 100), + "evidence_id": _bounded(value.get("evidence_id"), 255), + "tenant_id": _bounded(value.get("tenant_id"), 36), + "version": _bounded(value.get("version"), 255), + "checksum": _bounded(value.get("checksum"), 255), + "source_ref": _bounded(value.get("source_ref"), 2_000), + "derived_from": [_bounded(item, 2_000) for item in derived], + "responsible_actor_ref": _bounded( + value.get("responsible_actor_ref"), + 255, + ), + "captured_at": _bounded(value.get("captured_at"), 100), + } + + +def _bounded(value: object, limit: int) -> str | None: + return str(value)[:limit] if value is not None else None + + +def _sensitive_key(value: str) -> bool: + normalized = value.strip().casefold().replace("-", "_") + return any( + part in normalized + for part in ( + "authorization", + "cookie", + "credential", + "password", + "secret", + "token", + ) + ) + + +def _iso(value: datetime | None) -> str | None: + aware = _aware(value) + return aware.isoformat() if aware else None + + +def _aware(value: datetime | None) -> datetime | None: + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Payments DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "payments" or record.module_id != "payments": + raise ValueError("Payments DSAR cannot plan a foreign provider record.") + if record.resource_type not in {"payment_obligation"} | _ATTRIBUTION_TYPES: + raise ValueError("Payments DSAR record type is invalid.") + if not record.resource_id: + raise ValueError("Payments DSAR record identity is incomplete.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "payments" or action.module_id != "payments": + raise ValueError("Payments DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("payments:"): + raise ValueError("Payments DSAR action identity is invalid.") + + +__all__ = ["PAYMENTS_DSAR_CAPABILITY", "PaymentsDsarProvider"] diff --git a/src/govoplan_payments/backend/manifest.py b/src/govoplan_payments/backend/manifest.py index 1515916..4b86dd7 100644 --- a/src/govoplan_payments/backend/manifest.py +++ b/src/govoplan_payments/backend/manifest.py @@ -26,6 +26,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_payments.backend.db import models as payment_models +from govoplan_payments.backend.dsar_provider import ( + PAYMENTS_DSAR_CAPABILITY, + PaymentsDsarProvider, +) from govoplan_payments.backend.service import SqlPaymentRequestProvider @@ -62,6 +66,10 @@ def _payment_requests(_context: ModuleContext) -> SqlPaymentRequestProvider: return SqlPaymentRequestProvider() +def _dsar_provider(_context: ModuleContext) -> PaymentsDsarProvider: + return PaymentsDsarProvider() + + def _tenant_summary(session: object, tenant_id: str) -> dict[str, int]: if not hasattr(session, "query"): return {"payment_requests": 0, "paid_payments": 0, "reconciliations": 0} @@ -191,14 +199,26 @@ manifest = ModuleManifest( ), provides_interfaces=( ModuleInterfaceProvider(name=CAPABILITY_PAYMENT_REQUESTS, version="1.0.0"), + ModuleInterfaceProvider(name=PAYMENTS_DSAR_CAPABILITY, version="0.1.0"), ), - capability_factories={CAPABILITY_PAYMENT_REQUESTS: _payment_requests}, + capability_factories={ + CAPABILITY_PAYMENT_REQUESTS: _payment_requests, + PAYMENTS_DSAR_CAPABILITY: _dsar_provider, + }, capability_documentation={ CAPABILITY_PAYMENT_REQUESTS: CapabilityDocumentation( label="Payment request and reconciliation", summary="Creates replay-safe obligations and records exact, evidence-bound manual settlement.", contract_version="1.0.0", ), + PAYMENTS_DSAR_CAPABILITY: CapabilityDocumentation( + label="Payments data-subject request provider", + summary=( + "Exports exact verified payment evidence or minimized operator " + "attribution with retain-only erasure outcomes." + ), + contract_version="0.1.0", + ), }, route_factory=_router, migration_spec=MigrationSpec( @@ -227,6 +247,58 @@ manifest = ModuleManifest( ), tenant_summary_providers=(_tenant_summary,), documentation=( + DocumentationTopic( + id="payments.data-subject-requests", + title="Payment data-subject requests", + summary=( + "Export exact payment obligations and financial evidence without " + "using payment descriptions as an identity search surface." + ), + body=( + "Payments has no resident or applicant identity column and does not " + "search payment subjects, context JSON, metadata, or source records " + "for a person. Full financial access therefore requires an exact " + "payment row id, payment id, or human payment reference supplied as " + "a verified external subject reference. The resulting package contains " + "the obligation amount, currency, subject, status, dates, source and " + "bounded context references, reconciliation facts and typed evidence " + "references, and lifecycle-event facts. Reconciliation metadata, event " + "payloads, hashes, replay keys, provider data, inspection URLs, and " + "credentials are excluded. A request containing only an account, " + "membership, identity, or exact actor reference receives minimized " + "request, reconciliation, and event attribution for that operator; it " + "does not expose payment subjects. Every result is exact-tenant and " + "bounded. All erasure actions are retain-only and non-executable because " + "obligations, settlements, evidence links, and lifecycle attribution " + "remain governed financial and statutory evidence." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("data_subject", "operator", "auditor", "module_admin"), + related_modules=("core", "cases", "workflow_engine", "ledger"), + metadata={ + "help_contexts": [ + "payments.workspace", + "payments.state.requested", + "payments.state.paid", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_exact_payment": ( + "Returns the obligation and bounded financial evidence for a " + "verified exact payment identifier." + ), + "export_operator_attribution": ( + "Returns minimized financial activity, never arbitrary payment " + "content." + ), + "retain_payment_evidence": ( + "Keeps financial evidence under configured statutory retention " + "and legal hold." + ), + }, + }, + ), DocumentationTopic( id="payments.requests-and-reconciliation", title="Payment requests and manual reconciliation", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..a5b69c6 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,418 @@ +from __future__ import annotations + +import json +import unittest +from datetime import UTC, datetime + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarProvider, + DsarRecordRef, + DsarSubjectRef, +) +from govoplan_core.db.base import Base +from govoplan_core.privacy.dsar_workflow import ( + create_data_subject_request, + search_data_subject_request, +) +from govoplan_payments.backend.db.models import ( + PaymentEvent, + PaymentObligation, + PaymentReconciliation, +) +from govoplan_payments.backend.dsar_provider import ( + PAYMENTS_DSAR_CAPABILITY, + PaymentsDsarProvider, +) +from govoplan_payments.backend.manifest import manifest + + +NOW = datetime(2026, 8, 21, 10, 0, tzinfo=UTC) + + +class _Registry: + def __init__(self, provider: PaymentsDsarProvider, *, active: bool = True) -> None: + self.provider = provider + self.active = active + + def capability_names(self): + return (PAYMENTS_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "payments" + + def tenant_entitlement_resolver(self): + active = self.active + + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("payments",) if active else ()}, + )() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + self._assert_capability(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "payments"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != PAYMENTS_DSAR_CAPABILITY: + raise KeyError(name) + + +class PaymentsDsarProviderTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(self.engine) + self.session = Session(self.engine) + self.provider = PaymentsDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + self._seed() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _obligation( + self, + row_id: str, + *, + tenant_id: str = "tenant-1", + payment_id: str, + payment_reference: str, + requested_by_ref: str, + subject: str, + ) -> PaymentObligation: + return PaymentObligation( + id=row_id, + tenant_id=tenant_id, + payment_id=payment_id, + payment_reference=payment_reference, + source_module="cases", + source_resource_type="case", + source_resource_id=f"case-{row_id}", + amount_minor=12_500, + currency="EUR", + subject=subject, + status="paid", + idempotency_key=f"idempotency-{row_id}-do-not-export", + request_sha256="a" * 64, + requested_at=NOW, + requested_by_ref=requested_by_ref, + settled_at=NOW, + context_refs={ + "service": "resident-permit", + "access_token": "payment-secret-do-not-export", + }, + details={"private": "obligation-metadata-do-not-export"}, + ) + + def _seed(self) -> None: + subject_payment = self._obligation( + "payment-row-1", + payment_id="payment-1", + payment_reference="PAY-0001", + requested_by_ref="account:account-1", + subject="Resident permit fee", + ) + other_payment = self._obligation( + "payment-row-other", + payment_id="payment-other", + payment_reference="PAY-OTHER", + requested_by_ref="account:account-other", + subject="Other person's private payment", + ) + other_tenant = self._obligation( + "payment-row-other-tenant", + tenant_id="tenant-2", + payment_id="payment-other-tenant", + payment_reference="PAY-TENANT-2", + requested_by_ref="account:account-1", + subject="Other tenant private payment", + ) + self.session.add_all((subject_payment, other_payment, other_tenant)) + self.session.flush() + self.session.add_all( + ( + PaymentReconciliation( + id="reconciliation-1", + tenant_id="tenant-1", + reconciliation_id="reconciliation-command-1", + payment_row_id="payment-row-1", + mode="manual_full", + amount_minor=12_500, + currency="EUR", + transaction_reference="BANK-REFERENCE-1", + evidence_ref={ + "kind": "record", + "owner_module": "records", + "evidence_id": "record-1", + "tenant_id": "tenant-1", + "version": "4", + "checksum": "b" * 64, + "source_ref": "records:record-1:v4", + "derived_from": ["bank-statement-1"], + "responsible_actor_ref": "account:account-1", + "captured_at": NOW.isoformat(), + "inspection_url": "/records/record-1", + }, + idempotency_key="reconcile-key-do-not-export", + request_sha256="c" * 64, + received_at=NOW, + recorded_at=NOW, + recorded_by_ref="account:account-1", + details={"private": "reconciliation-metadata-do-not-export"}, + ), + PaymentEvent( + id="event-1", + tenant_id="tenant-1", + event_id="payment-event-1", + payment_row_id="payment-row-1", + event_type="payment.reconciled", + status="paid", + occurred_at=NOW, + actor_ref="account:account-1", + payload={"secret": "event-payload-do-not-export"}, + ), + PaymentEvent( + id="event-other", + tenant_id="tenant-1", + event_id="payment-event-other", + payment_row_id="payment-row-other", + event_type="payment.requested", + status="requested", + occurred_at=NOW, + actor_ref="account:account-other", + payload={"private": "other event"}, + ), + ) + ) + + def test_exact_payment_reference_exports_bounded_financial_evidence(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"payments.reference": "PAY-0001"} + ), + ) + + self.assertEqual(["payment-row-1"], [record.resource_id for record in records]) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("Resident permit fee", exported) + self.assertIn("BANK-REFERENCE-1", exported) + self.assertIn("records:record-1:v4", exported) + self.assertIn("payment.reconciled", exported) + self.assertIn("[redacted]", exported) + for excluded in ( + "payment-secret-do-not-export", + "obligation-metadata-do-not-export", + "reconciliation-metadata-do-not-export", + "event-payload-do-not-export", + "idempotency-payment-row-1-do-not-export", + "reconcile-key-do-not-export", + "inspection_url", + "Other person's private payment", + "Other tenant private payment", + ): + self.assertNotIn(excluded, exported) + + def test_actor_search_is_minimized_and_does_not_expose_subject(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="account-1"), + ) + + self.assertEqual( + { + "payment_request_attribution", + "payment_reconciliation_attribution", + "payment_event_attribution", + }, + {record.resource_type for record in records}, + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("PAY-0001", exported) + self.assertIn("recorded_payment_reconciliation", exported) + self.assertNotIn("Resident permit fee", exported) + self.assertNotIn("BANK-REFERENCE-1", exported) + self.assertNotIn("Other person's private payment", exported) + self.assertNotIn("Other tenant private payment", exported) + + def test_identifiers_corroborate_and_alias_conflicts_fail_closed(self) -> None: + corroborated = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={ + "payments.payment": "payment-1", + "payments.reference": "PAY-0001", + } + ), + ) + mismatched = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={ + "payments.payment": "payment-1", + "payments.reference": "PAY-OTHER", + } + ), + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"payments.account": "account-other"}, + ), + ) + + self.assertEqual(["payment-row-1"], [item.resource_id for item in corroborated]) + self.assertEqual((), mismatched) + self.assertEqual((), conflict) + + def test_erasure_is_retain_only_and_foreign_inputs_are_rejected(self) -> None: + subject = DsarSubjectRef(external_references={"payments.reference": "PAY-0001"}) + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=subject, + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=records, + ) + self.assertTrue(all(action.kind == "retain" for action in actions)) + self.assertTrue(all(not action.executable for action in actions)) + results = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + self.assertTrue(all(result.status == "blocked" for result in results)) + self.assertIsNotNone(self.session.get(PaymentObligation, "payment-row-1")) + + with self.assertRaisesRegex(ValueError, "foreign provider record"): + self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=( + DsarRecordRef( + provider_id="ledger", + module_id="ledger", + resource_type="payment_obligation", + resource_id="payment-row-1", + category="financial", + title="Foreign payment", + ), + ), + ) + with self.assertRaisesRegex(ValueError, "foreign provider action"): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=( + DsarErasureActionRef( + action_id="ledger:retain:payment:payment-row-1", + provider_id="ledger", + module_id="ledger", + kind="retain", + resource_type="payment_obligation", + resource_id="payment-row-1", + title="Retain payment", + rationale="Financial evidence", + executable=False, + ), + ), + request_id="dsar-1", + ) + + def test_core_workflow_and_manifest_register_provider(self) -> None: + subject = DsarSubjectRef(external_references={"payments.reference": "PAY-0001"}) + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-PAYMENTS-1", + request_kind="access", + subject=subject, + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=row, + expected_revision=1, + ) + self.assertEqual( + [PAYMENTS_DSAR_CAPABILITY], row.coverage["provider_capabilities"] + ) + self.assertEqual(1, row.search_result["record_count"]) + + inactive = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-PAYMENTS-2", + request_kind="access", + subject=subject, + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider, active=False), + row=inactive, + expected_revision=1, + ) + self.assertEqual( + [PAYMENTS_DSAR_CAPABILITY], + inactive.coverage["inactive_provider_capabilities"], + ) + + self.assertIn(PAYMENTS_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn(PAYMENTS_DSAR_CAPABILITY, manifest.capability_documentation) + self.assertIn( + PAYMENTS_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + self.assertTrue( + any( + topic.id == "payments.data-subject-requests" + and {"admin", "user"}.issubset(topic.documentation_types) + for topic in manifest.documentation + ) + ) + + +if __name__ == "__main__": + unittest.main()