581 lines
20 KiB
Python
581 lines
20 KiB
Python
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()
|