from __future__ import annotations import json import unittest from datetime import UTC, datetime, timedelta from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.dsar import DsarProvider, 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_risk_compliance.backend.db.models import ( RiskAssuranceEdge, RiskAssuranceNode, RiskSanctionsEntry, RiskSanctionsListSnapshot, RiskScreeningCandidate, RiskScreeningDisposition, RiskScreeningException, RiskScreeningRun, RiskScreeningSubjectSnapshot, ) from govoplan_risk_compliance.backend.dsar_provider import ( RISK_COMPLIANCE_DSAR_CAPABILITY, RiskComplianceDsarProvider, ) from govoplan_risk_compliance.backend.manifest import manifest NOW = datetime(2026, 8, 22, 10, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: RiskComplianceDsarProvider) -> None: self.provider = provider def capability_names(self): return (RISK_COMPLIANCE_DSAR_CAPABILITY,) def capability_owner(self, name): if name != RISK_COMPLIANCE_DSAR_CAPABILITY: raise KeyError(name) return "risk_compliance" def tenant_entitlement_resolver(self): class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type("State", (), {"effective_modules": ("risk_compliance",)})() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs if name != RISK_COMPLIANCE_DSAR_CAPABILITY: raise KeyError(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "risk_compliance"})(),) class RiskComplianceDsarProviderTests(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 = RiskComplianceDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: list_snapshot = RiskSanctionsListSnapshot( id="list-1", tenant_id="tenant-1", visibility="tenant", connector_snapshot_ref="connector:snapshot:secret-do-not-export", provider_id="un", publisher="United Nations", jurisdiction="global", list_type="sanctions", source_id="consolidated", source_version="2026-08-22", publication_at=NOW, effective_at=NOW, acquired_at=NOW, sha256="list-sha-do-not-export", connector_run_id="connector-run-do-not-export", raw_evidence_ref="evidence-ref-do-not-export", source_parser_version="parser-v1", normalization_version="normalizer-v1", signature_evidence={"secret": "signature-do-not-export"}, provenance={"secret": "provenance-do-not-export"}, entry_count=1, status="active", imported_by="account-1", imported_at=NOW, created_at=NOW, updated_at=NOW, ) subject_snapshot = RiskScreeningSubjectSnapshot( id="subject-snapshot-1", tenant_id="tenant-1", subject_ref="party:person-1", subject_type="person", primary_name="Ada Example", normalized_name="ada example normalized-do-not-export", aliases=["Ada E."], identifiers=[ { "type": "resident-number", "value": "resident-123", "unexpected": "identifier-extra-do-not-export", } ], dates=["1990-01-01"], addresses=[ { "street": "Example Street 1", "city": "Exampletown", "country": "DE", "unexpected": "address-extra-do-not-export", } ], fingerprint="subject-fingerprint-do-not-export", submitted_by="account-1", created_at=NOW, updated_at=NOW, ) run = RiskScreeningRun( id="screening-1", tenant_id="tenant-1", subject_snapshot_id="subject-snapshot-1", list_snapshot_id="list-1", idempotency_key="screening-idempotency-do-not-export", request_hash="screening-request-hash-do-not-export", matcher_version="matcher-v1", normalization_version="normalizer-v1", policy_version="policy-v1", policy_snapshot={"secret": "policy-snapshot-do-not-export"}, status="complete", outcome="review", candidate_count=1, started_at=NOW, completed_at=NOW, created_by="account-1", created_at=NOW, updated_at=NOW, ) sanctions_entry = RiskSanctionsEntry( id="entry-1", snapshot_id="list-1", source_entry_id="third-party-entry-do-not-export", subject_type="person", primary_name="Third Party Name Do Not Export", normalized_name="third party", original_script_name=None, reference_number="third-party-reference-do-not-export", listed_on=None, programmes=[], measures=[], raw_evidence_locator="third-party-evidence-do-not-export", details={"secret": "third-party-details-do-not-export"}, created_at=NOW, updated_at=NOW, ) candidate = RiskScreeningCandidate( id="candidate-1", tenant_id="tenant-1", run_id="screening-1", entry_id="entry-1", score=91, match_kind="fuzzy", evidence=[{"secret": "candidate-evidence-do-not-export"}], review_status="confirmed", current_disposition_id="disposition-1", created_at=NOW, updated_at=NOW, ) disposition = RiskScreeningDisposition( id="disposition-1", tenant_id="tenant-1", candidate_id="candidate-1", decision="false_positive", reason="review-reason-do-not-export", evidence_refs=["review-evidence-do-not-export"], scope="subject_entry", expires_at=NOW + timedelta(days=30), review_at=NOW + timedelta(days=15), actor_account_id="account-1", actor_membership_id="membership-1", actor_authority={"secret": "authority-do-not-export"}, separation_status="independent", override_reason="override-reason-do-not-export", created_at=NOW, ) exception = RiskScreeningException( id="exception-1", tenant_id="tenant-1", subject_fingerprint="exception-fingerprint-do-not-export", source_entry_ref="exception-source-entry-do-not-export", scope="subject_entry", status="active", reason="exception-reason-do-not-export", evidence_refs=["exception-evidence-do-not-export"], starts_at=NOW, expires_at=NOW + timedelta(days=30), review_at=NOW + timedelta(days=15), originating_disposition_id="disposition-1", created_by="account-1", created_at=NOW, updated_at=NOW, ) node = RiskAssuranceNode( id="node-1", tenant_id="tenant-1", stable_id="control-1", kind="control", revision=1, label="Sensitive control label do not export", description="Sensitive control description do not export", state="active", owner_ref="owner-secret-do-not-export", scope_ref="scope-secret-do-not-export", governed_object_ref="object-secret-do-not-export", valid_from=NOW, recorded_at=NOW, provenance={"secret": "node-provenance-do-not-export"}, legal_basis_refs=["legal-secret-do-not-export"], policy_refs=["policy-secret-do-not-export"], evidence_refs=["node-evidence-do-not-export"], classification="restricted", created_by="account-1", created_at=NOW, updated_at=NOW, ) edge = RiskAssuranceEdge( id="edge-1", tenant_id="tenant-1", stable_id="relation-1", revision=1, source_node_ref="control-1", target_node_ref="risk-1", relation="mitigates", state="active", owner_ref="edge-owner-secret-do-not-export", scope_ref="edge-scope-secret-do-not-export", valid_from=NOW, recorded_at=NOW, provenance={"secret": "edge-provenance-do-not-export"}, legal_basis_refs=["edge-legal-secret-do-not-export"], policy_refs=["edge-policy-secret-do-not-export"], evidence_refs=["edge-evidence-do-not-export"], created_by="account-1", created_at=NOW, updated_at=NOW, ) other_tenant_node = RiskAssuranceNode( id="node-other", tenant_id="tenant-2", stable_id="other-control", kind="control", revision=1, label="Other tenant", state="active", owner_ref="owner", valid_from=NOW, recorded_at=NOW, provenance={}, legal_basis_refs=[], policy_refs=[], evidence_refs=[], classification="internal", created_by="account-1", ) self.session.add_all( ( list_snapshot, subject_snapshot, run, sanctions_entry, candidate, disposition, exception, node, edge, other_tenant_node, ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef( account_id="account-1", membership_id="membership-1", external_references={ "risk_compliance.subject": "party:person-1", }, ) def test_search_exports_subject_data_and_minimized_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual( { "screening_subject_submission", "screening_actor_attribution", "snapshot_import_actor_attribution", "disposition_actor_attribution", "exception_actor_attribution", "assurance_node_actor_attribution", "assurance_edge_actor_attribution", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("Ada Example", exported) self.assertIn("resident-123", exported) for excluded in ( "normalized-do-not-export", "identifier-extra-do-not-export", "address-extra-do-not-export", "subject-fingerprint-do-not-export", "screening-idempotency-do-not-export", "screening-request-hash-do-not-export", "policy-snapshot-do-not-export", "Third Party Name Do Not Export", "third-party-reference-do-not-export", "candidate-evidence-do-not-export", "review-reason-do-not-export", "review-evidence-do-not-export", "authority-do-not-export", "override-reason-do-not-export", "exception-fingerprint-do-not-export", "exception-source-entry-do-not-export", "exception-reason-do-not-export", "Sensitive control label do not export", "node-provenance-do-not-export", "edge-owner-secret-do-not-export", "edge-evidence-do-not-export", "list-sha-do-not-export", "signature-do-not-export", ): self.assertNotIn(excluded, exported) def test_subject_data_requires_exact_module_reference(self) -> None: account_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="account-1"), ) reference_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "risk_compliance.subject": "party:person-1", } ), ) self.assertNotIn( "screening_subject_submission", {record.resource_type for record in account_only}, ) self.assertEqual( {"screening_subject_submission"}, {record.resource_type for record in reference_only}, ) def test_conflicts_narrowing_and_tenant_boundaries(self) -> None: conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={ "risk_compliance.account": "account-other", }, ), ) narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={ "risk_compliance.subject": "party:person-1", "risk_compliance.screening": "screening-1", }, ), ) full = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual((), conflict) self.assertEqual( { "screening_subject_submission", "screening_actor_attribution", "disposition_actor_attribution", }, {record.resource_type for record in narrowed}, ) self.assertNotIn("node-other", {record.resource_id for record in full}) def test_erasure_retains_legal_and_accountability_evidence(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self._subject(), records=records, ) self.assertTrue(actions) self.assertTrue( all(action.kind == "retain" and not action.executable for action in actions) ) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self._subject(), actions=actions, request_id="dsar-risk-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) def test_manifest_and_core_workflow_discover_provider(self) -> None: self.assertIn(RISK_COMPLIANCE_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn( "risk_compliance.data-subject-requests", {topic.id for topic in manifest.documentation}, ) row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-RISK-1", request_kind="access", subject=self._subject(), purpose="Risk screening access request", legal_basis=None, due_at=None, requested_by_account_id="operator-1", ) search_data_subject_request( self.session, registry=_Registry(self.provider), row=row, expected_revision=row.resource_revision, ) self.assertEqual("searched", row.status) self.assertEqual(7, row.search_result["record_count"]) if __name__ == "__main__": unittest.main()