from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db.models import Account, Group, User from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef from govoplan_core.db.base import Base from govoplan_core.privacy.dsar_workflow import ( DataSubjectRequest, create_data_subject_request, execute_data_subject_erasure, plan_data_subject_erasure, search_data_subject_request, ) from govoplan_scheduling.backend.db.models import ( SchedulingCandidateSlot, SchedulingNotification, SchedulingParticipant, SchedulingPublicEnrollmentLink, SchedulingRequest, ) from govoplan_scheduling.backend.dsar_provider import ( SCHEDULING_DSAR_CAPABILITY, SchedulingDsarProvider, ) from govoplan_scheduling.backend.manifest import manifest class _Registry: def __init__( self, provider: SchedulingDsarProvider, *, scheduling_active: bool = True, ) -> None: self.provider = provider self.scheduling_active = scheduling_active def capability_names(self): return (SCHEDULING_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "scheduling" def tenant_entitlement_resolver(self): scheduling_active = self.scheduling_active class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type( "State", (), { "effective_modules": ( ("scheduling",) if scheduling_active else () ) }, )() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs self._assert_capability(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "scheduling"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != SCHEDULING_DSAR_CAPABILITY: raise KeyError(name) class SchedulingDsarProviderTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:", future=True) Base.metadata.create_all( bind=self.engine, tables=[ Account.__table__, User.__table__, Group.__table__, ChangeSequenceEntry.__table__, DataSubjectRequest.__table__, SchedulingRequest.__table__, SchedulingPublicEnrollmentLink.__table__, SchedulingCandidateSlot.__table__, SchedulingParticipant.__table__, SchedulingNotification.__table__, ], ) self.session = sessionmaker(bind=self.engine, future=True)() now = datetime.now(timezone.utc) account = Account( id="account-1", email="subject@example.test", normalized_email="subject@example.test", display_name="Subject", ) other_account = Account( id="account-2", email="other@example.test", normalized_email="other@example.test", display_name="Other", ) self.user = User( id="membership-1", tenant_id="tenant-1", account_id=account.id, email="subject@example.test", display_name="Subject", ) other_user = User( id="membership-2", tenant_id="tenant-1", account_id=other_account.id, email="other@example.test", display_name="Other", ) self.request = SchedulingRequest( id="request-active", tenant_id="tenant-1", title="Choose an appointment", description="Scheduling context visible to the participant", location="Town hall", status="collecting", poll_id="poll-id-do-not-export", organizer_user_id=other_user.id, deadline_at=now + timedelta(days=3), anonymous_password_protection_enabled=True, anonymous_password_hash="password-hash-do-not-export", calendar_integration_enabled=True, calendar_id="calendar-id-do-not-export", calendar_hold_enabled=True, calendar_event_id="calendar-event-id-do-not-export", metadata_={"secret": "request-metadata-do-not-export"}, ) slot = SchedulingCandidateSlot( id="slot-subject", tenant_id="tenant-1", request_id=self.request.id, poll_option_id="poll-option-id-do-not-export", label="Tuesday morning", description="First option", start_at=now + timedelta(days=1), end_at=now + timedelta(days=1, hours=1), timezone="Europe/Berlin", location="Town hall", position=0, freebusy_checked_at=now, freebusy_status="busy", freebusy_conflicts=[{"person": "Unrelated conflict person do not export"}], tentative_hold_event_id="hold-event-id-do-not-export", metadata_={"secret": "slot-metadata-do-not-export"}, ) self.engaged = SchedulingParticipant( id="participant-engaged", tenant_id="tenant-1", request_id=self.request.id, respondent_id=self.user.id, display_name="Subject Person", email="Subject@Example.Test", participant_type="internal", required=True, status="responded", poll_invitation_id="poll-invitation-id-do-not-export", participation_gateway="public-gateway-do-not-export", self_enrollment_proof_hash="proof-hash-do-not-export", bound_account_id=account.id, account_bound_at=now, last_invited_at=now, responded_at=now, response_comment="Subject response comment", metadata_={"secret": "participant-metadata-do-not-export"}, ) self.unengaged = SchedulingParticipant( id="participant-unengaged", tenant_id="tenant-1", request_id=self.request.id, respondent_id=self.user.id, display_name="Subject duplicate draft", email=None, participant_type="external", required=False, status="draft", metadata_={"directory": "internal-directory-data-do-not-export"}, ) unrelated = SchedulingParticipant( id="participant-other", tenant_id="tenant-1", request_id=self.request.id, respondent_id=other_user.id, display_name="Unrelated Person", email="other@example.test", status="responded", poll_invitation_id="other-invitation-do-not-export", responded_at=now, response_comment="Unrelated response do not export", ) notification = SchedulingNotification( id="notification-subject", tenant_id="tenant-1", request_id=self.request.id, participant_id=self.engaged.id, event_kind="invitation", channel="mail", recipient="subject@example.test", status="sent", payload={ "private": "notification-payload-do-not-export", "token": "notification-token-do-not-export", }, error="notification-error-do-not-export", sent_at=now, metadata_={"secret": "notification-metadata-do-not-export"}, ) unrelated_notification = SchedulingNotification( id="notification-other", tenant_id="tenant-1", request_id=self.request.id, participant_id=unrelated.id, event_kind="decision", channel="mail", recipient="other@example.test", status="sent", payload={"private": "other-notification-do-not-export"}, sent_at=now, ) organizer_request = SchedulingRequest( id="request-organized", tenant_id="tenant-1", title="Subject organized meeting", status="draft", poll_id="organizer-poll-do-not-export", organizer_user_id=self.user.id, ) organizer_slot = SchedulingCandidateSlot( id="slot-organized", tenant_id="tenant-1", request_id=organizer_request.id, label="Organizer option", start_at=now + timedelta(days=2), end_at=now + timedelta(days=2, hours=1), ) organizer_other_participant = SchedulingParticipant( id="participant-organizer-other", tenant_id="tenant-1", request_id=organizer_request.id, display_name="Organizer unrelated invitee do not export", email="organizer-other@example.test", status="draft", ) unrelated_request = SchedulingRequest( id="request-other", tenant_id="tenant-1", title="Unrelated request do not export", status="collecting", poll_id="unrelated-poll-do-not-export", organizer_user_id=other_user.id, ) tenant_two_request = SchedulingRequest( id="request-tenant-2", tenant_id="tenant-2", title="Tenant two request do not export", status="collecting", poll_id="tenant-two-poll-do-not-export", ) tenant_two_participant = SchedulingParticipant( id="participant-tenant-2", tenant_id="tenant-2", request_id=tenant_two_request.id, display_name="Tenant two subject", email="subject@example.test", status="draft", ) self.session.add_all( [ account, other_account, self.user, other_user, self.request, slot, self.engaged, self.unengaged, unrelated, notification, unrelated_notification, organizer_request, organizer_slot, organizer_other_participant, unrelated_request, tenant_two_request, tenant_two_participant, ] ) self.session.commit() self.provider = SchedulingDsarProvider() self.subject = DsarSubjectRef( account_id=account.id, membership_id=self.user.id, email="subject@example.test", ) def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_manifest_publishes_protocol_conforming_provider(self) -> None: provided_names = {item.name for item in manifest.provides_interfaces} self.assertIn(SCHEDULING_DSAR_CAPABILITY, provided_names) provider = manifest.capability_factories[SCHEDULING_DSAR_CAPABILITY](None) self.assertIsInstance(provider, DsarProvider) self.assertIn( "scheduling.privacy.data-subject-requests", {topic.id for topic in manifest.documentation}, ) def test_search_is_tenant_scoped_minimized_and_participant_specific(self) -> None: records = self._records() resource_types = {record.resource_type for record in records} self.assertTrue( { "scheduling_request", "scheduling_candidate_slot", "scheduling_participant", "scheduling_notification", }.issubset(resource_types) ) self.assertEqual( {"participant-engaged", "participant-unengaged"}, { record.resource_id for record in records if record.resource_type == "scheduling_participant" }, ) engaged = next( record for record in records if record.resource_id == "participant-engaged" ) self.assertEqual("Subject response comment", engaged.data["response_comment"]) serialized = repr([record.to_dict() for record in records]) for hidden in ( "participant-other", "Unrelated Person", "other@example.test", "Unrelated response do not export", "notification-other", "other-notification-do-not-export", "participant-organizer-other", "Organizer unrelated invitee do not export", "request-other", "Unrelated request do not export", "request-tenant-2", "Tenant two request do not export", "participant-tenant-2", "poll-id-do-not-export", "password-hash-do-not-export", "calendar-id-do-not-export", "calendar-event-id-do-not-export", "request-metadata-do-not-export", "poll-option-id-do-not-export", "Unrelated conflict person do not export", "hold-event-id-do-not-export", "slot-metadata-do-not-export", "poll-invitation-id-do-not-export", "public-gateway-do-not-export", "proof-hash-do-not-export", "participant-metadata-do-not-export", "internal-directory-data-do-not-export", "notification-payload-do-not-export", "notification-token-do-not-export", "notification-error-do-not-export", "notification-metadata-do-not-export", "organizer-poll-do-not-export", ): self.assertNotIn(hidden, serialized) def test_conflicting_email_references_fail_closed_for_participant_data( self, ) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( email="subject@example.test", external_references={"scheduling.email": "other@example.test"}, ), ) self.assertEqual((), records) def test_plan_retains_evidence_and_only_anonymizes_unengaged_participant( self, ) -> None: actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self.subject, records=self._records(), ) self.assertTrue({"retain", "manual_review"}.issubset({a.kind for a in actions})) self.assertTrue( any( action.action_id == "scheduling:retain:scheduling_participant:participant-engaged" for action in actions ) ) self.assertEqual( {"scheduling:anonymize:scheduling_participant:participant-unengaged"}, {action.action_id for action in actions if action.executable}, ) def test_execution_is_revalidated_tenant_bound_and_idempotent(self) -> None: action = self._anonymize_action() wrong_tenant = self.provider.execute_erasure( self.session, tenant_id="tenant-2", subject=self.subject, actions=(action,), request_id="dsar-wrong-tenant", ) self.assertEqual("blocked", wrong_tenant[0].status) first = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=(action,), request_id="dsar-scheduling-1", ) self.assertEqual("executed", first[0].status) self.session.flush() self.assertEqual("removed", self.unengaged.status) self.assertIsNotNone(self.unengaged.deleted_at) self.assertIsNone(self.unengaged.display_name) self.assertIsNone(self.unengaged.email) self.assertIsNone(self.unengaged.respondent_id) self.assertIsNone(self.unengaged.metadata_) repeated = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=(action,), request_id="dsar-scheduling-1", ) self.assertEqual("unchanged", repeated[0].status) def test_execution_blocks_when_evidence_appears_after_planning(self) -> None: action = self._anonymize_action() self.session.add( SchedulingNotification( id="notification-late", tenant_id="tenant-1", request_id=self.request.id, participant_id=self.unengaged.id, event_kind="invitation", recipient=self.unengaged.email, status="pending", payload={}, ) ) self.session.flush() result = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=(action,), request_id="dsar-stale", ) self.assertEqual("blocked", result[0].status) self.assertIsNone(self.unengaged.deleted_at) def test_core_workflow_discovers_active_provider_and_skips_it_when_disabled( self, ) -> None: request = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-SCHEDULING-1", request_kind="access_and_erasure", subject=self.subject, purpose="Respond to an authorized privacy request.", legal_basis="Article 15 and 17 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) self.session.commit() registry = _Registry(self.provider) search_data_subject_request( self.session, registry=registry, row=request, expected_revision=1, ) self.assertEqual("searched", request.status) self.assertEqual(["scheduling"], request.coverage["covered_modules"]) plan_data_subject_erasure( self.session, registry=registry, row=request, expected_revision=2, ) executable_ids = [ action["action_id"] for action in request.erasure_plan["actions"] if action["executable"] ] execute_data_subject_erasure( self.session, registry=registry, row=request, expected_revision=3, action_ids=executable_ids, ) self.assertEqual("completed", request.status) disabled = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-SCHEDULING-DISABLED", request_kind="access", subject=self.subject, purpose="Verify disabled-module coverage.", legal_basis="Article 15 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) search_data_subject_request( self.session, registry=_Registry(self.provider, scheduling_active=False), row=disabled, expected_revision=1, ) self.assertEqual(0, disabled.search_result["record_count"]) self.assertEqual( [SCHEDULING_DSAR_CAPABILITY], disabled.coverage["inactive_provider_capabilities"], ) def _records(self): return self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) def _anonymize_action(self): return next( action for action in self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self.subject, records=self._records(), ) if action.executable ) if __name__ == "__main__": unittest.main()