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_poll.backend.db.models import ( Poll, PollInvitation, PollLifecycleTransition, PollParticipationSubmission, PollResponse, ) from govoplan_poll.backend.dsar_provider import POLL_DSAR_CAPABILITY, PollDsarProvider from govoplan_poll.backend.manifest import manifest NOW = datetime(2026, 8, 21, 16, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: PollDsarProvider) -> None: self.provider = provider def capability_names(self): return (POLL_DSAR_CAPABILITY,) def capability_owner(self, name): if name != POLL_DSAR_CAPABILITY: raise KeyError(name) return "poll" def tenant_entitlement_resolver(self): class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type("State", (), {"effective_modules": ("poll",)})() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs if name != POLL_DSAR_CAPABILITY: raise KeyError(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "poll"})(),) class PollDsarProviderTests(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 = PollDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: poll = Poll( id="poll-1", tenant_id="tenant-1", slug="resident-availability", title="Resident appointment availability", description="Institutional description", kind="availability", status="open", visibility="private", result_visibility="after_close", allow_anonymous=True, allow_response_update=True, min_choices=1, created_by_user_id="account-1", metadata_={"secret": "poll-metadata-do-not-export"}, ) self.session.add(poll) self.session.flush() invitation = PollInvitation( id="invitation-1", tenant_id="tenant-1", poll_id="poll-1", token_hash="token-hash-do-not-export", respondent_id="account-1", respondent_label="Ada Example", email="Ada@Example.DE", expires_at=NOW, last_used_at=NOW, response_gateway_={"secret": "gateway-do-not-export"}, participation_policy_={"secret": "policy-do-not-export"}, metadata_={"secret": "invitation-metadata-do-not-export"}, ) response = PollResponse( id="response-1", tenant_id="tenant-1", poll_id="poll-1", respondent_id="account-1", respondent_label="Ada Example", answers=[{"option_id": "option-a", "available": True}], submitted_at=NOW, metadata_={"secret": "response-metadata-do-not-export"}, ) anonymous = PollResponse( id="response-anonymous", tenant_id="tenant-1", poll_id="poll-1", respondent_id=None, answers=[{"private": "anonymous-answer-do-not-correlate"}], submitted_at=NOW, ) other = PollResponse( id="response-other", tenant_id="tenant-1", poll_id="poll-1", respondent_id="account-other", respondent_label="Other Person", answers=[{"private": "other-answer-do-not-export"}], submitted_at=NOW, ) self.session.add_all((invitation, response, anonymous, other)) self.session.flush() self.session.add( PollParticipationSubmission( id="submission-1", tenant_id="tenant-1", poll_id="poll-1", invitation_id="invitation-1", response_id="response-1", idempotency_key="submission-idempotency-do-not-export", request_fingerprint="submission-fingerprint-do-not-export", ) ) self.session.add( PollLifecycleTransition( id="transition-1", tenant_id="tenant-1", poll_id="poll-1", action="open", from_status="draft", to_status="open", idempotency_key="transition-idempotency-do-not-export", actor_user_id="account-1", metadata_={"secret": "transition-metadata-do-not-export"}, ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef(account_id="account-1", email="ada@example.de") def test_search_exports_identified_participation_and_minimized_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual( { "poll_response", "poll_invitation", "poll_creator_attribution", "poll_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("Ada@Example.DE", exported) for excluded in ( "token-hash-do-not-export", "gateway-do-not-export", "policy-do-not-export", "invitation-metadata-do-not-export", "response-metadata-do-not-export", "submission-idempotency-do-not-export", "submission-fingerprint-do-not-export", "transition-idempotency-do-not-export", "transition-metadata-do-not-export", "anonymous-answer-do-not-correlate", "other-answer-do-not-export", ): self.assertNotIn(excluded, exported) def test_email_only_follows_explicit_invitation_response_link(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(email="ADA@EXAMPLE.DE"), ) self.assertEqual( {"poll_invitation", "poll_response"}, {record.resource_type for record in records}, ) def test_poll_narrowing_conflicts_and_anonymous_limit(self) -> None: narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"poll.poll": "poll-1"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"poll.respondent": "account-other"}, ), ) poll_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(external_references={"poll.poll": "poll-1"}), ) self.assertTrue(narrowed) self.assertEqual((), conflict) self.assertEqual((), poll_only) self.assertNotIn( "response-anonymous", {record.resource_id for record in narrowed} ) def test_erasure_requires_review_and_preserves_results(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.assertEqual( {"manual_review", "retain"}, {action.kind for action in actions} ) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self._subject(), actions=actions, request_id="dsar-poll-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) self.assertIsNone(self.session.get(PollResponse, "response-1").deleted_at) def test_manifest_and_core_workflow_discover_provider(self) -> None: self.assertIn(POLL_DSAR_CAPABILITY, manifest.capability_factories) row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-POLL-1", request_kind="access", subject=self._subject(), purpose="Poll 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(4, row.search_result["record_count"]) if __name__ == "__main__": unittest.main()