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 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_voting.backend.db.models import ( VotingBallotRevision, VotingCastRecord, VotingConfidentialBallot, VotingConfidentialCast, VotingLifecycleEvent, ) from govoplan_voting.backend.dsar_provider import ( VOTING_DSAR_CAPABILITY, VotingDsarProvider, ) from govoplan_voting.backend.manifest import manifest NOW = datetime(2026, 8, 21, 15, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: VotingDsarProvider) -> None: self.provider = provider def capability_names(self): return (VOTING_DSAR_CAPABILITY,) def capability_owner(self, name): if name != VOTING_DSAR_CAPABILITY: raise KeyError(name) return "voting" def tenant_entitlement_resolver(self): class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type("State", (), {"effective_modules": ("voting",)})() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs if name != VOTING_DSAR_CAPABILITY: raise KeyError(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "voting"})(),) class VotingDsarProviderTests(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 = VotingDsarProvider() 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( VotingBallotRevision( id="ballot-revision-1", tenant_id="tenant-1", ballot_id="ballot-1", revision=1, state="closed", assurance_profile="recorded", method="single_choice", definition_sha256="definition-hash-do-not-export", electorate_sha256="electorate-hash-do-not-export", recorded_at=NOW, payload={ "electorate": "ballot-electorate-do-not-export", "options": "ballot-options-do-not-export", }, created_by="account-1", ) ) self.session.add_all( ( VotingCastRecord( id="recorded-cast-1", tenant_id="tenant-1", ballot_id="ballot-1", definition_sha256="cast-definition-hash-do-not-export", elector_id="account-1", generation=1, selections=["option-a"], weight=1, cast_at=NOW, idempotency_key="cast-idempotency-do-not-export", receipt_sha256="recorded-receipt-1", actor_id="account-1", ), VotingCastRecord( id="recorded-cast-other", tenant_id="tenant-1", ballot_id="ballot-1", definition_sha256="other-definition", elector_id="account-other", generation=1, selections=["private-other-selection-do-not-export"], weight=1, cast_at=NOW, idempotency_key="other-idempotency", receipt_sha256="other-receipt", actor_id="account-other", ), VotingCastRecord( id="recorded-cast-proxy", tenant_id="tenant-1", ballot_id="ballot-1", definition_sha256="proxy-definition", elector_id="account-proxy-subject", generation=1, selections=["proxy-selection-do-not-export"], weight=1, cast_at=NOW, idempotency_key="proxy-idempotency-do-not-export", receipt_sha256="proxy-receipt-do-not-export", actor_id="account-1", ), ) ) confidential_ballot = VotingConfidentialBallot( id="confidential-ballot-row", tenant_id="tenant-1", provider_ballot_ref="provider-ballot-1", ballot_id="ballot-confidential", definition_sha256="confidential-definition-do-not-export", electorate_sha256="confidential-electorate-hash-do-not-export", assurance_profile="confidential", method="single_choice", state="closed", options=[{"private": "confidential-options-do-not-export"}], electorate=[{"private": "confidential-electorate-do-not-export"}], allow_replacement=True, quorum_weight=1, threshold_numerator=1, threshold_denominator=2, vault_id="vault-do-not-export", preparation_idempotency_key="preparation-idempotency-do-not-export", preparation_request_sha256="preparation-hash-do-not-export", prepared_at=NOW, result={"private": "confidential-result-do-not-export"}, ) self.session.add(confidential_ballot) self.session.flush() self.session.add( VotingConfidentialCast( id="confidential-cast-1", tenant_id="tenant-1", provider_ballot_id="confidential-ballot-row", elector_id="account-1", generation=1, definition_sha256="confidential-cast-definition-do-not-export", ciphertext=b"ciphertext-do-not-export", encryption_envelope_id="envelope-do-not-export", encryption_resource_id="resource-key-do-not-export", weight=1, cast_at=NOW, idempotency_key="confidential-idempotency-do-not-export", request_sha256="confidential-request-hash-do-not-export", receipt_sha256="confidential-receipt-1", ) ) self.session.add( VotingLifecycleEvent( id="lifecycle-1", tenant_id="tenant-1", ballot_id="ballot-1", sequence=1, event_type="ballot.opened", recorded_at=NOW, actor_id="account-1", payload={"secret": "lifecycle-payload-do-not-export"}, ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef(account_id="account-1") def test_search_separates_recorded_and_confidential_disclosure(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual( { "recorded_ballot_cast", "recorded_cast_actor_attribution", "confidential_ballot_participation", "ballot_actor_attribution", "voting_lifecycle_actor_attribution", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("option-a", exported) self.assertIn("recorded-receipt-1", exported) self.assertIn("confidential-receipt-1", exported) self.assertIn('"selections_disclosed": false', exported) for excluded in ( "private-other-selection-do-not-export", "proxy-selection-do-not-export", "proxy-receipt-do-not-export", "ciphertext-do-not-export", "envelope-do-not-export", "resource-key-do-not-export", "confidential-options-do-not-export", "confidential-electorate-do-not-export", "confidential-result-do-not-export", "ballot-electorate-do-not-export", "lifecycle-payload-do-not-export", "cast-idempotency-do-not-export", ): self.assertNotIn(excluded, exported) def test_ballot_narrowing_and_conflicting_elector_fail_closed(self) -> None: narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"voting.ballot": "ballot-confidential"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"voting.elector": "account-other"}, ), ) ballot_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"voting.ballot": "ballot-1"} ), ) self.assertEqual( ["confidential_ballot_participation"], [record.resource_type for record in narrowed], ) self.assertEqual((), conflict) self.assertEqual((), ballot_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(actions) 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-voting-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) self.assertEqual(3, self.session.query(VotingCastRecord).count()) def test_manifest_and_core_workflow_discover_provider(self) -> None: self.assertIn(VOTING_DSAR_CAPABILITY, manifest.capability_factories) row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-VOTING-1", request_kind="access", subject=self._subject(), purpose="Voting participation 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(5, row.search_result["record_count"]) if __name__ == "__main__": unittest.main()