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()