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_committee.backend.db.models import ( CommitteeDecisionProjection, CommitteeWorkspaceEvent, CommitteeWorkspaceRevision, ) from govoplan_committee.backend.dsar_provider import ( COMMITTEE_DSAR_CAPABILITY, CommitteeDsarProvider, ) from govoplan_committee.backend.manifest import manifest 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, ) NOW = datetime(2026, 8, 21, 17, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: CommitteeDsarProvider) -> None: self.provider = provider def capability_names(self): return (COMMITTEE_DSAR_CAPABILITY,) def capability_owner(self, name): if name != COMMITTEE_DSAR_CAPABILITY: raise KeyError(name) return "committee" def tenant_entitlement_resolver(self): class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type("State", (), {"effective_modules": ("committee",)})() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs if name != COMMITTEE_DSAR_CAPABILITY: raise KeyError(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "committee"})(),) class CommitteeDsarProviderTests(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 = CommitteeDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: self.session.add_all( ( CommitteeWorkspaceRevision( id="workspace-revision-1", tenant_id="tenant-1", object_kind="meeting", object_id="meeting-1", revision=1, parent_kind="body", parent_id="body-1", state="scheduled", title="Private meeting title do not export", search_text="private search text do not export", recorded_at=NOW, payload={"minutes": "private minutes do not export"}, changed_by="account-1", ), CommitteeWorkspaceRevision( id="workspace-revision-other", tenant_id="tenant-1", object_kind="meeting", object_id="meeting-other", revision=1, state="scheduled", title="Other meeting", search_text="other search text", recorded_at=NOW, payload={"private": "other payload"}, changed_by="account-other", ), ) ) self.session.add( CommitteeWorkspaceEvent( id="event-1", tenant_id="tenant-1", object_kind="meeting", object_id="meeting-1", object_revision=1, event_id="event-external-1", event_type="meeting.scheduled", occurred_at=NOW, actor_id="account-1", idempotency_key="event-idempotency-do-not-export", request_sha256="event-request-hash-do-not-export", payload={"secret": "event-payload-do-not-export"}, ) ) self.session.add( CommitteeDecisionProjection( id="decision-projection-1", tenant_id="tenant-1", decision_id="decision-1", revision="1", meeting_id="meeting-1", agenda_item_id="agenda-1", state="effective", recorded_at=NOW, payload={ "reasoning": "decision-reasoning-do-not-export", "ballot": "individual-ballot-do-not-export", }, changed_by="account-1", ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef(account_id="account-1") def test_search_exports_minimized_committee_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual( { "committee_workspace_actor_attribution", "committee_event_actor_attribution", "committee_decision_actor_attribution", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("meeting-1", exported) self.assertIn("agenda-1", exported) for excluded in ( "Private meeting title do not export", "private search text do not export", "private minutes do not export", "event-idempotency-do-not-export", "event-request-hash-do-not-export", "event-payload-do-not-export", "decision-reasoning-do-not-export", "individual-ballot-do-not-export", "meeting-other", ): self.assertNotIn(excluded, exported) def test_object_narrowing_and_actor_conflicts_fail_closed(self) -> None: narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={ "committee.object_kind": "meeting", "committee.object": "meeting-1", }, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"committee.actor": "account-other"}, ), ) object_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"committee.object": "meeting-1"} ), ) self.assertEqual( { "committee_workspace_actor_attribution", "committee_event_actor_attribution", "committee_decision_actor_attribution", }, {record.resource_type for record in narrowed}, ) self.assertEqual((), conflict) self.assertEqual((), object_only) def test_erasure_is_retain_only(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(all(action.kind == "retain" for action in actions)) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self._subject(), actions=actions, request_id="dsar-committee-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) self.assertEqual(2, self.session.query(CommitteeWorkspaceRevision).count()) def test_manifest_and_core_workflow_discover_provider(self) -> None: self.assertIn(COMMITTEE_DSAR_CAPABILITY, manifest.capability_factories) row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-COMMITTEE-1", request_kind="access", subject=self._subject(), purpose="Committee attribution 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(3, row.search_result["record_count"]) if __name__ == "__main__": unittest.main()