From 7e03fe62acabfd294110fe482e00245b990446c8 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 12:15:56 +0200 Subject: [PATCH] feat(voting): add governed DSAR coverage --- src/govoplan_voting/backend/dsar_provider.py | 481 +++++++++++++++++++ src/govoplan_voting/backend/manifest.py | 62 ++- tests/test_dsar_provider.py | 315 ++++++++++++ 3 files changed, 857 insertions(+), 1 deletion(-) create mode 100644 src/govoplan_voting/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_voting/backend/dsar_provider.py b/src/govoplan_voting/backend/dsar_provider.py new file mode 100644 index 0000000..7989548 --- /dev/null +++ b/src/govoplan_voting/backend/dsar_provider.py @@ -0,0 +1,481 @@ +from __future__ import annotations + +from collections.abc import 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_voting.backend.db.models import ( + VotingBallotRevision, + VotingCastRecord, + VotingConfidentialBallot, + VotingConfidentialCast, + VotingLifecycleEvent, +) + + +VOTING_DSAR_CAPABILITY = dsar_capability_name("voting") +_MAX_RECORDS = 5_000 +_MAX_SELECTIONS = 1_000 +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + elector_ids: tuple[str, ...] + actor_ids: tuple[str, ...] + ballot_id: str | None + + +class VotingDsarProvider: + provider_id = "voting" + module_id = "voting" + + 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 () + records: list[DsarRecordRef] = [] + + personal_casts = db.query(VotingCastRecord).filter( + VotingCastRecord.tenant_id == tenant_id, + VotingCastRecord.elector_id.in_(selectors.elector_ids), + ) + actor_casts = db.query(VotingCastRecord).filter( + VotingCastRecord.tenant_id == tenant_id, + VotingCastRecord.actor_id.in_(selectors.actor_ids), + ~VotingCastRecord.elector_id.in_(selectors.elector_ids), + ) + ballots = db.query(VotingBallotRevision).filter( + VotingBallotRevision.tenant_id == tenant_id, + VotingBallotRevision.created_by.in_(selectors.actor_ids), + ) + events = db.query(VotingLifecycleEvent).filter( + VotingLifecycleEvent.tenant_id == tenant_id, + VotingLifecycleEvent.actor_id.in_(selectors.actor_ids), + ) + confidential = ( + db.query(VotingConfidentialCast, VotingConfidentialBallot) + .join( + VotingConfidentialBallot, + VotingConfidentialCast.provider_ballot_id + == VotingConfidentialBallot.id, + ) + .filter( + VotingConfidentialCast.tenant_id == tenant_id, + VotingConfidentialBallot.tenant_id == tenant_id, + VotingConfidentialCast.elector_id.in_(selectors.elector_ids), + ) + ) + if selectors.ballot_id: + personal_casts = personal_casts.filter( + VotingCastRecord.ballot_id == selectors.ballot_id + ) + actor_casts = actor_casts.filter( + VotingCastRecord.ballot_id == selectors.ballot_id + ) + ballots = ballots.filter( + VotingBallotRevision.ballot_id == selectors.ballot_id + ) + events = events.filter( + VotingLifecycleEvent.ballot_id == selectors.ballot_id + ) + confidential = confidential.filter( + VotingConfidentialBallot.ballot_id == selectors.ballot_id + ) + + records.extend( + _recorded_cast(row) + for row in _limited( + personal_casts, + VotingCastRecord.cast_at, + VotingCastRecord.id, + label="recorded cast", + ) + ) + records.extend( + _cast_actor_attribution(row) + for row in _limited( + actor_casts, + VotingCastRecord.cast_at, + VotingCastRecord.id, + label="cast actor attribution", + ) + ) + confidential_rows = ( + confidential.order_by( + VotingConfidentialCast.cast_at, + VotingConfidentialCast.id, + ) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(confidential_rows) > _MAX_RECORDS: + raise ValueError( + "Voting DSAR confidential-cast limit exceeded; narrow selectors." + ) + records.extend( + _confidential_cast_receipt(cast, ballot) + for cast, ballot in confidential_rows + ) + records.extend( + _ballot_actor_attribution(row) + for row in _limited( + ballots, + VotingBallotRevision.recorded_at, + VotingBallotRevision.id, + label="ballot attribution", + ) + ) + records.extend( + _event_actor_attribution(row) + for row in _limited( + events, + VotingLifecycleEvent.recorded_at, + VotingLifecycleEvent.id, + label="lifecycle attribution", + ) + ) + if len(records) > _MAX_RECORDS: + raise ValueError( + "Voting DSAR combined result limit exceeded; narrow the selectors." + ) + return tuple( + sorted(records, key=lambda item: (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("Voting DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + actions.append( + DsarErasureActionRef( + action_id=( + f"voting: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 "Ballot evidence must retain integrity.", + 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("Voting DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "retain": + raise ValueError("Voting DSAR publishes retain-only actions.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Ballot participation and lifecycle evidence remains unchanged " + "to preserve integrity, certification, and challenge history." + ), + 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("voting.account"), + references.get("access.account"), + ), + "membership_id": _coalesce( + subject.membership_id, + references.get("voting.membership"), + references.get("tenancy.membership"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("voting.identity"), + references.get("identity.id"), + ), + "elector_id": _coalesce( + references.get("voting.elector"), + references.get("voting.elector_id"), + ), + "actor_id": _coalesce( + references.get("voting.actor"), + references.get("voting.created_by"), + ), + "ballot_id": _coalesce( + references.get("voting.ballot"), + references.get("voting.ballot_id"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + base_ids = tuple( + dict.fromkeys( + value + for value in ( + _optional_string(values["account_id"]), + _prefixed("account", values["account_id"]), + _optional_string(values["membership_id"]), + _prefixed("membership", values["membership_id"]), + _optional_string(values["identity_id"]), + _prefixed("identity", values["identity_id"]), + ) + if value + ) + ) + direct_elector = _optional_string(values["elector_id"]) + direct_actor = _optional_string(values["actor_id"]) + for direct in (direct_elector, direct_actor): + if direct and base_ids and direct not in base_ids: + return None + if not base_ids and direct_elector and direct_actor and direct_elector != direct_actor: + return None + elector_ids = base_ids or ((direct_elector or direct_actor,) if direct_elector or direct_actor else ()) + actor_ids = base_ids or ((direct_actor or direct_elector,) if direct_actor or direct_elector else ()) + if not elector_ids: + return None + return _SubjectSelectors( + elector_ids=elector_ids, + actor_ids=actor_ids, + ballot_id=_optional_string(values["ballot_id"]), + ) + + +def _recorded_cast(row: VotingCastRecord) -> DsarRecordRef: + if not isinstance(row.selections, list) or len(row.selections) > _MAX_SELECTIONS: + raise ValueError("Voting recorded selections exceed the DSAR bound.") + return DsarRecordRef( + provider_id="voting", + module_id="voting", + resource_type="recorded_ballot_cast", + resource_id=row.id, + category="identified_recorded_vote", + title="Recorded ballot participation", + data={ + "ballot_id": row.ballot_id, + "generation": row.generation, + "selections": [str(item)[:255] for item in row.selections], + "weight": row.weight, + "cast_at": _iso(row.cast_at), + "superseded_at": _iso(row.superseded_at), + "receipt_sha256": row.receipt_sha256, + "assurance": "recorded_and_reconstructable", + }, + observed_at=_aware(row.cast_at), + immutable_evidence=True, + retention_reason=( + "Recorded votes are attributable, reconstructable ballot evidence." + ), + ) + + +def _cast_actor_attribution(row: VotingCastRecord) -> DsarRecordRef: + return DsarRecordRef( + provider_id="voting", + module_id="voting", + resource_type="recorded_cast_actor_attribution", + resource_id=row.id, + category="ballot_operator_attribution", + title="Recorded cast actor attribution", + data={ + "ballot_id": row.ballot_id, + "generation": row.generation, + "cast_at": _iso(row.cast_at), + "superseded_at": _iso(row.superseded_at), + "activity": "recorded_cast_for_elector", + }, + observed_at=_aware(row.cast_at), + immutable_evidence=True, + retention_reason="Cast actor attribution is immutable ballot evidence.", + ) + + +def _confidential_cast_receipt( + row: VotingConfidentialCast, + ballot: VotingConfidentialBallot, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="voting", + module_id="voting", + resource_type="confidential_ballot_participation", + resource_id=row.id, + category="confidential_vote_participation_receipt", + title="Confidential ballot participation", + data={ + "ballot_id": ballot.ballot_id, + "provider_ballot_ref": ballot.provider_ballot_ref, + "assurance_profile": ballot.assurance_profile, + "method": ballot.method, + "ballot_state": ballot.state, + "generation": row.generation, + "weight": row.weight, + "cast_at": _iso(row.cast_at), + "superseded_at": _iso(row.superseded_at), + "receipt_sha256": row.receipt_sha256, + "selections_disclosed": False, + }, + observed_at=_aware(row.cast_at), + immutable_evidence=True, + retention_reason=( + "Confidential participation receipts are retained without ciphertext or " + "selection disclosure." + ), + ) + + +def _ballot_actor_attribution(row: VotingBallotRevision) -> DsarRecordRef: + return DsarRecordRef( + provider_id="voting", + module_id="voting", + resource_type="ballot_actor_attribution", + resource_id=row.id, + category="ballot_governance_attribution", + title="Ballot revision actor attribution", + data={ + "ballot_id": row.ballot_id, + "revision": row.revision, + "state": row.state, + "assurance_profile": row.assurance_profile, + "method": row.method, + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "activity": "recorded_ballot_revision", + }, + observed_at=_aware(row.recorded_at), + immutable_evidence=True, + retention_reason="Ballot revision attribution is governance evidence.", + ) + + +def _event_actor_attribution(row: VotingLifecycleEvent) -> DsarRecordRef: + return DsarRecordRef( + provider_id="voting", + module_id="voting", + resource_type="voting_lifecycle_actor_attribution", + resource_id=row.id, + category="ballot_governance_attribution", + title="Voting lifecycle actor attribution", + data={ + "ballot_id": row.ballot_id, + "sequence": row.sequence, + "event_type": row.event_type, + "recorded_at": _iso(row.recorded_at), + }, + observed_at=_aware(row.recorded_at), + immutable_evidence=True, + retention_reason="Voting lifecycle attribution is governance evidence.", + ) + + +def _limited(query, first, second, *, label: str): + rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all() + if len(rows) > _MAX_RECORDS: + raise ValueError(f"Voting DSAR {label} limit exceeded; narrow selectors.") + return rows + + +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 _prefixed(prefix: str, value: object) -> str | None: + normalized = _optional_string(value) + return f"{prefix}:{normalized}" if normalized else None + + +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("Voting DSAR requires a SQLAlchemy Session.") + return value + + +_RESOURCE_TYPES = { + "recorded_ballot_cast", + "recorded_cast_actor_attribution", + "confidential_ballot_participation", + "ballot_actor_attribution", + "voting_lifecycle_actor_attribution", +} + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "voting" or record.module_id != "voting": + raise ValueError("Voting DSAR cannot plan a foreign provider record.") + if record.resource_type not in _RESOURCE_TYPES or not record.resource_id: + raise ValueError("Voting DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "voting" or action.module_id != "voting": + raise ValueError("Voting DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("voting:retain:"): + raise ValueError("Voting DSAR action identity is invalid.") + + +__all__ = ["VOTING_DSAR_CAPABILITY", "VotingDsarProvider"] diff --git a/src/govoplan_voting/backend/manifest.py b/src/govoplan_voting/backend/manifest.py index daa8c9e..67b85b1 100644 --- a/src/govoplan_voting/backend/manifest.py +++ b/src/govoplan_voting/backend/manifest.py @@ -38,6 +38,10 @@ from govoplan_core.core.voting import ( ) from govoplan_core.db.base import Base from govoplan_voting.backend.db import models as voting_models +from govoplan_voting.backend.dsar_provider import ( + VOTING_DSAR_CAPABILITY, + VotingDsarProvider, +) from govoplan_voting.backend.service import SqlVotingBallots from govoplan_voting.backend.local_confidential_provider import ( LOCAL_CONFIDENTIAL_PROVIDER_ID, @@ -96,6 +100,10 @@ def _local_confidential_provider( return LocalConfidentialVotingProvider(context.registry) +def _dsar_provider(_context: ModuleContext) -> VotingDsarProvider: + return VotingDsarProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: current = session.query(voting_models.VotingBallotRevision).filter( voting_models.VotingBallotRevision.tenant_id == tenant_id, @@ -125,6 +133,7 @@ manifest = ModuleManifest( name=voting_provider_capability(LOCAL_CONFIDENTIAL_PROVIDER_ID), version="0.1.0", ), + ModuleInterfaceProvider(name=VOTING_DSAR_CAPABILITY, version="0.1.0"), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -264,13 +273,22 @@ manifest = ModuleManifest( voting_provider_capability( LOCAL_CONFIDENTIAL_PROVIDER_ID ): _local_confidential_provider, + VOTING_DSAR_CAPABILITY: _dsar_provider, }, capability_documentation={ CAPABILITY_VOTING_BALLOTS: CapabilityDocumentation( label="Governed ballots", summary="Creates frozen electorates, records eligible votes, closes deterministic tallies, and certifies aggregate results.", contract_version="0.1.0", - ) + ), + VOTING_DSAR_CAPABILITY: CapabilityDocumentation( + label="Voting data-subject request provider", + summary=( + "Exports identified recorded votes, confidential participation " + "receipts, and minimized actor attribution without weakening secrecy." + ), + contract_version="0.1.0", + ), }, migration_spec=MigrationSpec( module_id=MODULE_ID, @@ -301,6 +319,48 @@ manifest = ModuleManifest( ), tenant_summary_providers=(_tenant_summary,), documentation=( + DocumentationTopic( + id="voting.data-subject-requests", + title="Voting data-subject requests", + summary=( + "Distinguish reconstructable recorded votes from confidential " + "participation when exporting a subject's ballot data." + ), + body=( + "Voting correlates exact tenant and elector identifiers and can narrow " + "an already verified search to one ballot. Recorded ballots are " + "explicitly attributable and reconstructable, so a subject receives " + "their own bounded selections, weight, generation, timestamps, and " + "receipt. Confidential ballots return participation, assurance, " + "generation, timing, and receipt metadata only. Ciphertext, encryption " + "envelopes, resource-key references, electorate payloads, and choices " + "are never disclosed. Acting on another elector's cast produces only " + "minimized actor attribution. Ballot identifiers alone reveal no " + "personal participation. All voting records are retained to preserve " + "integrity, certification, recount, and challenge evidence." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + related_modules=("core", "committee", "identity_trust", "encryption"), + metadata={ + "help_contexts": [ + "voting.ballot", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_recorded_vote": ( + "Returns the subject's reconstructable recorded selections." + ), + "export_confidential_receipt": ( + "Returns participation metadata without choices or ciphertext." + ), + "retain_ballot_evidence": ( + "Preserves ballot integrity and challenge history." + ), + }, + }, + ), DocumentationTopic( id="voting.assurance", title="Voting assurance and certification", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..358e28b --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,315 @@ +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()