From 02ec5423b91ceb31ac776fa19c135ad4331305b0 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 12:15:56 +0200 Subject: [PATCH] feat(poll): add governed DSAR coverage --- src/govoplan_poll/backend/dsar_provider.py | 506 +++++++++++++++++++++ src/govoplan_poll/backend/manifest.py | 63 ++- tests/test_dsar_provider.py | 284 ++++++++++++ tests/test_manifest.py | 2 +- 4 files changed, 853 insertions(+), 2 deletions(-) create mode 100644 src/govoplan_poll/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_poll/backend/dsar_provider.py b/src/govoplan_poll/backend/dsar_provider.py new file mode 100644 index 0000000..06e6ffa --- /dev/null +++ b/src/govoplan_poll/backend/dsar_provider.py @@ -0,0 +1,506 @@ +from __future__ import annotations + +import json +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy import func, or_ +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_poll.backend.db.models import ( + Poll, + PollInvitation, + PollLifecycleTransition, + PollParticipationSubmission, + PollResponse, +) + + +POLL_DSAR_CAPABILITY = dsar_capability_name("poll") +_MAX_RECORDS = 5_000 +_MAX_ANSWERS = 1_000 +_MAX_ANSWER_BYTES = 256 * 1024 +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + respondent_ids: tuple[str, ...] + actor_ids: tuple[str, ...] + email: str | None + poll_id: str | None + invitation_id: str | None + response_id: str | None + + +class PollDsarProvider: + provider_id = "poll" + module_id = "poll" + + 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] = [] + + invitation_conditions = [] + if selectors.respondent_ids: + invitation_conditions.append( + PollInvitation.respondent_id.in_(selectors.respondent_ids) + ) + if selectors.email: + invitation_conditions.append( + func.lower(PollInvitation.email) == selectors.email + ) + invitations = db.query(PollInvitation).filter( + PollInvitation.tenant_id == tenant_id, + or_(*invitation_conditions), + ) + if selectors.poll_id: + invitations = invitations.filter(PollInvitation.poll_id == selectors.poll_id) + if selectors.invitation_id: + invitations = invitations.filter(PollInvitation.id == selectors.invitation_id) + invitation_rows = _limited( + invitations, + PollInvitation.created_at, + PollInvitation.id, + label="invitation", + ) + records.extend(_invitation_record(row) for row in invitation_rows) + + linked_response_ids: tuple[str, ...] = () + if invitation_rows: + invitation_ids = [row.id for row in invitation_rows] + linked_rows = ( + db.query(PollParticipationSubmission.response_id) + .filter( + PollParticipationSubmission.tenant_id == tenant_id, + PollParticipationSubmission.invitation_id.in_(invitation_ids), + ) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(linked_rows) > _MAX_RECORDS: + raise ValueError( + "Poll DSAR participation-link limit exceeded; narrow selectors." + ) + linked_response_ids = tuple( + dict.fromkeys(str(response_id) for (response_id,) in linked_rows) + ) + + response_conditions = [] + if selectors.invitation_id: + if linked_response_ids: + response_conditions.append(PollResponse.id.in_(linked_response_ids)) + else: + if selectors.respondent_ids: + response_conditions.append( + PollResponse.respondent_id.in_(selectors.respondent_ids) + ) + if linked_response_ids: + response_conditions.append(PollResponse.id.in_(linked_response_ids)) + if response_conditions: + responses = db.query(PollResponse).filter( + PollResponse.tenant_id == tenant_id, + or_(*response_conditions), + ) + if selectors.poll_id: + responses = responses.filter(PollResponse.poll_id == selectors.poll_id) + if selectors.response_id: + responses = responses.filter(PollResponse.id == selectors.response_id) + records.extend( + _response_record(row) + for row in _limited( + responses, + PollResponse.submitted_at, + PollResponse.id, + label="response", + ) + ) + + if selectors.actor_ids: + polls = db.query(Poll).filter( + Poll.tenant_id == tenant_id, + Poll.created_by_user_id.in_(selectors.actor_ids), + ) + transitions = db.query(PollLifecycleTransition).filter( + PollLifecycleTransition.tenant_id == tenant_id, + PollLifecycleTransition.actor_user_id.in_(selectors.actor_ids), + ) + if selectors.poll_id: + polls = polls.filter(Poll.id == selectors.poll_id) + transitions = transitions.filter( + PollLifecycleTransition.poll_id == selectors.poll_id + ) + records.extend( + _creator_attribution(row) + for row in _limited( + polls, + Poll.created_at, + Poll.id, + label="creator attribution", + ) + ) + records.extend( + _transition_attribution(row) + for row in _limited( + transitions, + PollLifecycleTransition.created_at, + PollLifecycleTransition.id, + label="lifecycle attribution", + ) + ) + + if len(records) > _MAX_RECORDS: + raise ValueError( + "Poll 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("Poll DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + participation = record.resource_type in { + "poll_response", + "poll_invitation", + } + actions.append( + DsarErasureActionRef( + action_id=( + f"poll:{'manual_review' if participation else 'retain'}:" + f"{record.resource_type}:{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind="manual_review" if participation else "retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=("Review " if participation else "Retain ") + record.title, + rationale=( + "Removing or anonymizing participation may change published " + "results, response-update behavior, or retained invitation " + "evidence and therefore requires the Poll owner and retention " + "authority to review the effect." + if participation + else record.retention_reason + or "Poll actor attribution remains governance evidence." + ), + 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("Poll DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind not in {"manual_review", "retain"}: + raise ValueError("Poll DSAR publishes non-executable actions only.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Poll participation remains unchanged pending result and " + "retention-impact review." + if action.kind == "manual_review" + else "Poll lifecycle attribution remains governance evidence." + ), + 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("poll.account"), + references.get("access.account"), + ), + "membership_id": _coalesce( + subject.membership_id, + references.get("poll.membership"), + references.get("tenancy.membership"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("poll.identity"), + references.get("identity.id"), + ), + "respondent_id": _coalesce( + references.get("poll.respondent"), + references.get("poll.respondent_id"), + ), + "email": _coalesce_email(subject.email, references.get("poll.email")), + "poll_id": _coalesce( + references.get("poll.poll"), references.get("poll.poll_id") + ), + "invitation_id": _coalesce( + references.get("poll.invitation"), + references.get("poll.invitation_id"), + ), + "response_id": _coalesce( + references.get("poll.response"), + references.get("poll.response_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_respondent = _optional_string(values["respondent_id"]) + if direct_respondent and base_ids and direct_respondent not in base_ids: + return None + respondent_ids = base_ids or ((direct_respondent,) if direct_respondent else ()) + email = _optional_string(values["email"]) + if not respondent_ids and not email: + return None + return _SubjectSelectors( + respondent_ids=respondent_ids, + actor_ids=base_ids, + email=email, + poll_id=_optional_string(values["poll_id"]), + invitation_id=_optional_string(values["invitation_id"]), + response_id=_optional_string(values["response_id"]), + ) + + +def _response_record(row: PollResponse) -> DsarRecordRef: + answers = _answers(row.answers) + return DsarRecordRef( + provider_id="poll", + module_id="poll", + resource_type="poll_response", + resource_id=row.id, + category="identified_poll_participation", + title=f"Poll response: {row.poll.title[:500]}", + data={ + "poll_id": row.poll_id, + "poll_title": row.poll.title[:500], + "poll_kind": row.poll.kind, + "respondent_id": row.respondent_id, + "respondent_label": (row.respondent_label or "")[:500] or None, + "answers": answers, + "submitted_at": _iso(row.submitted_at), + "retired_at": _iso(row.deleted_at), + }, + observed_at=_aware(row.updated_at), + retention_reason=( + "Response erasure or anonymization requires Poll result and retention review." + ), + ) + + +def _invitation_record(row: PollInvitation) -> DsarRecordRef: + return DsarRecordRef( + provider_id="poll", + module_id="poll", + resource_type="poll_invitation", + resource_id=row.id, + category="poll_invitation_and_contact", + title=f"Poll invitation: {row.poll.title[:500]}", + data={ + "poll_id": row.poll_id, + "poll_title": row.poll.title[:500], + "respondent_id": row.respondent_id, + "respondent_label": (row.respondent_label or "")[:500] or None, + "email": row.email, + "expires_at": _iso(row.expires_at), + "revoked_at": _iso(row.revoked_at), + "last_used_at": _iso(row.last_used_at), + "created_at": _iso(row.created_at), + }, + observed_at=_aware(row.updated_at), + retention_reason=( + "Invitation erasure requires participation and response-link review." + ), + ) + + +def _creator_attribution(row: Poll) -> DsarRecordRef: + return DsarRecordRef( + provider_id="poll", + module_id="poll", + resource_type="poll_creator_attribution", + resource_id=row.id, + category="poll_governance_attribution", + title="Poll creator attribution", + data={ + "poll_id": row.id, + "kind": row.kind, + "status": row.status, + "visibility": row.visibility, + "opens_at": _iso(row.opens_at), + "closes_at": _iso(row.closes_at), + "created_at": _iso(row.created_at), + "activity": "created_poll", + }, + observed_at=_aware(row.created_at), + immutable_evidence=True, + retention_reason="Poll creator attribution is governance evidence.", + ) + + +def _transition_attribution(row: PollLifecycleTransition) -> DsarRecordRef: + return DsarRecordRef( + provider_id="poll", + module_id="poll", + resource_type="poll_lifecycle_actor_attribution", + resource_id=row.id, + category="poll_governance_attribution", + title="Poll lifecycle actor attribution", + data={ + "poll_id": row.poll_id, + "action": row.action, + "from_status": row.from_status, + "to_status": row.to_status, + "created_at": _iso(row.created_at), + }, + observed_at=_aware(row.created_at), + immutable_evidence=True, + retention_reason="Poll lifecycle attribution is governance evidence.", + ) + + +def _answers(value: object) -> list[object]: + if not isinstance(value, list) or len(value) > _MAX_ANSWERS: + raise ValueError("Poll response answers exceed the DSAR bound.") + try: + encoded = json.dumps(value, ensure_ascii=False, sort_keys=True).encode("utf-8") + except (TypeError, ValueError) as exc: + raise ValueError("Poll response answers are not JSON serializable.") from exc + if len(encoded) > _MAX_ANSWER_BYTES: + raise ValueError("Poll response answer payload exceeds the DSAR byte bound.") + return json.loads(encoded.decode("utf-8")) + + +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"Poll 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 _coalesce_email(*values: str | None) -> str | None | object: + normalized = { + str(value).strip().casefold() + 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("Poll DSAR requires a SQLAlchemy Session.") + return value + + +_RESOURCE_TYPES = { + "poll_response", + "poll_invitation", + "poll_creator_attribution", + "poll_lifecycle_actor_attribution", +} + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "poll" or record.module_id != "poll": + raise ValueError("Poll DSAR cannot plan a foreign provider record.") + if record.resource_type not in _RESOURCE_TYPES or not record.resource_id: + raise ValueError("Poll DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "poll" or action.module_id != "poll": + raise ValueError("Poll DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("poll:"): + raise ValueError("Poll DSAR action identity is invalid.") + + +__all__ = ["POLL_DSAR_CAPABILITY", "PollDsarProvider"] diff --git a/src/govoplan_poll/backend/manifest.py b/src/govoplan_poll/backend/manifest.py index 1aeca8e..ca4e3ee 100644 --- a/src/govoplan_poll/backend/manifest.py +++ b/src/govoplan_poll/backend/manifest.py @@ -5,6 +5,7 @@ from pathlib import Path from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard from govoplan_core.core.modules import ( + CapabilityDocumentation, DocumentationTopic, MigrationSpec, ModuleContext, @@ -18,6 +19,7 @@ from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING from govoplan_core.core.poll_participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY from govoplan_core.db.base import Base from govoplan_poll.backend.db import models as poll_models # noqa: F401 - populate Poll ORM metadata +from govoplan_poll.backend.dsar_provider import POLL_DSAR_CAPABILITY, PollDsarProvider MODULE_ID = "poll" MODULE_NAME = "Poll" @@ -144,6 +146,11 @@ def _poll_participation_gateway_provider(context: ModuleContext) -> object: return _poll_scheduling_provider(context) +def _dsar_provider(context: ModuleContext) -> PollDsarProvider: + del context + return PollDsarProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -159,6 +166,7 @@ manifest = ModuleManifest( ModuleInterfaceProvider(name="poll.workflow_context", version=MODULE_VERSION), ModuleInterfaceProvider(name="poll.signed_participation", version=MODULE_VERSION), ModuleInterfaceProvider(name="poll.governed_participation", version=MODULE_VERSION), + ModuleInterfaceProvider(name=POLL_DSAR_CAPABILITY, version="0.1.0"), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -168,6 +176,17 @@ manifest = ModuleManifest( capability_factories={ CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider, CAPABILITY_POLL_PARTICIPATION_GATEWAY: _poll_participation_gateway_provider, + POLL_DSAR_CAPABILITY: _dsar_provider, + }, + capability_documentation={ + POLL_DSAR_CAPABILITY: CapabilityDocumentation( + label="Poll data-subject request provider", + summary=( + "Exports identified responses, invitation contact data, and minimized " + "operator attribution without token or gateway secrets." + ), + contract_version="0.1.0", + ), }, migration_spec=MigrationSpec( module_id=MODULE_ID, @@ -196,7 +215,49 @@ manifest = ModuleManifest( label="Poll", ), ), - documentation=DOCUMENTATION, + documentation=( + DocumentationTopic( + id="poll.data-subject-requests", + title="Poll data-subject requests", + summary=( + "Export identified responses and invitations while preserving result " + "integrity and the boundary around anonymous participation." + ), + body=( + "Poll correlates exact respondent identifiers and normalized email " + "addresses inside the active tenant. A matching invitation can resolve " + "its explicitly linked participation submission and response without " + "exposing the signed token. Subject-owned responses include bounded " + "answers, respondent labels, Poll context, and retirement state. " + "Invitations include contact and lifecycle state but never token hashes, " + "gateway configuration, participation policy, metadata, fingerprints, " + "or idempotency values. Creator and lifecycle activity is exported only " + "as minimized attribution. Truly anonymous responses have no stable " + "subject selector and cannot be correlated. Participation erasure " + "requires manual result and retention review; no automatic action " + "silently changes a Poll outcome." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "participant", "organizer", "auditor"), + related_modules=("core", "scheduling", "notifications", "mail"), + metadata={ + "help_contexts": ["privacy.data-subject-requests"], + "consequence_classes": { + "export_identified_response": ( + "Returns bounded subject-owned answers and Poll context." + ), + "anonymous_limitation": ( + "Cannot correlate a response that deliberately has no subject identifier." + ), + "review_participation_erasure": ( + "Requires result-integrity and retention review." + ), + }, + }, + ), + *DOCUMENTATION, + ), architecture=declared_module_architecture( layer="communication_participation", kind="domain", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..3c42026 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,284 @@ +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() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 4e3c945..180dda2 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -25,7 +25,7 @@ class PollManifestTests(unittest.TestCase): self.assertIn("poll.signed_participation", {interface.name for interface in manifest.provides_interfaces}) self.assertIn("poll.governed_participation", {interface.name for interface in manifest.provides_interfaces}) self.assertIn(CAPABILITY_POLL_PARTICIPATION_GATEWAY, manifest.capability_factories) - self.assertEqual(manifest.version, "0.1.11") + self.assertEqual(manifest.version, "0.1.18") self.assertIn("poll:response:write", {permission.scope for permission in manifest.permissions})