from __future__ import annotations import unittest from datetime import datetime, timezone from unittest.mock import patch from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.policy import ( CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, PolicySourceStep, SchedulingParticipantPrivacyDecision, SchedulingParticipantPrivacyRequest, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.base import Base from govoplan_scheduling.backend.db.models import ( SchedulingCandidateSlot, SchedulingParticipant, SchedulingRequest, ) from govoplan_scheduling.backend.runtime import configure_runtime from govoplan_scheduling.backend.schemas import ( SchedulingCandidateSlotInput, SchedulingRequestCreateRequest, SchedulingRequestResponse, SchedulingRequestUpdateRequest, ) from govoplan_scheduling.backend.service import scheduling_request_response class _PrivacyPolicy: def __init__(self, effective_visibility: str) -> None: self.effective_visibility = effective_visibility self.requests: list[SchedulingParticipantPrivacyRequest] = [] def resolve_scheduling_participant_visibility( self, session: object, *, request: SchedulingParticipantPrivacyRequest, ) -> SchedulingParticipantPrivacyDecision: self.requests.append(request) return SchedulingParticipantPrivacyDecision( effective_visibility=self.effective_visibility, # type: ignore[arg-type] reason="Tenant participant privacy policy", source_path=( PolicySourceStep( scope_type="tenant", scope_id=request.tenant_id, label="Tenant", applied_fields=("scheduling_participant_visibility",), ), ), details={"provider_session_available": session is not None}, ) class SchedulingParticipantPrivacyTests(unittest.TestCase): def setUp(self) -> None: configure_runtime(registry=PlatformRegistry()) self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ SchedulingRequest.__table__, SchedulingCandidateSlot.__table__, SchedulingParticipant.__table__, ], ) self.Session = sessionmaker(bind=self.engine) self.session: Session = self.Session() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all( self.engine, tables=[ SchedulingParticipant.__table__, SchedulingCandidateSlot.__table__, SchedulingRequest.__table__, ], ) self.engine.dispose() configure_runtime(registry=PlatformRegistry()) def _request(self, *, participant_visibility: str | None = None) -> SchedulingRequest: values = { "tenant_id": "tenant-1", "title": "Privacy review", "status": "collecting", "poll_id": "poll-internal", "organizer_user_id": "organizer-1", "calendar_integration_enabled": True, "calendar_id": "calendar-internal", "calendar_freebusy_enabled": True, "calendar_hold_enabled": True, "create_calendar_event_on_decision": True, "calendar_event_id": "event-internal", "metadata_": {"connector_secret_ref": "secret-internal"}, } if participant_visibility is not None: values["participant_visibility"] = participant_visibility request = SchedulingRequest(**values) request.slots = [ SchedulingCandidateSlot( tenant_id="tenant-1", poll_option_id="poll-option-internal", label="Monday morning", start_at=datetime(2026, 7, 27, 9, tzinfo=timezone.utc), end_at=datetime(2026, 7, 27, 10, tzinfo=timezone.utc), timezone="Europe/Berlin", position=0, freebusy_checked_at=datetime(2026, 7, 21, 12, tzinfo=timezone.utc), freebusy_status="busy", freebusy_conflicts=[ { "calendar_id": "calendar-internal", "event_id": "event-internal", "uid": "connector-uid-internal", "recurrence_id": "recurrence-internal", "start_at": "2026-07-27T09:30:00+00:00", "end_at": "2026-07-27T09:45:00+00:00", "status": "CONFIRMED", }, {"error": "connector-internal failure"}, ], tentative_hold_event_id="hold-internal", metadata_={"provider_ref": "provider-internal"}, ) ] request.participants = [ SchedulingParticipant( tenant_id="tenant-1", respondent_id="alice-id", display_name="Alice", email="alice@example.test", participant_type="internal", status="responded", poll_invitation_id="invitation-alice-internal", last_invited_at=datetime(2026, 7, 20, 12, tzinfo=timezone.utc), responded_at=datetime(2026, 7, 21, 12, tzinfo=timezone.utc), metadata_={"private": "alice"}, ), SchedulingParticipant( tenant_id="tenant-1", respondent_id="bob-id", display_name="Bob", email="bob@example.test", participant_type="external", status="invited", metadata_={"private": "bob"}, ), ] self.session.add(request) self.session.flush() return request @staticmethod def _participant_projection(request: SchedulingRequest) -> dict[str, object]: return scheduling_request_response( request, actor_ids=("alice-id", "alice@example.test"), actor_user_id="alice-account", ) @staticmethod def _configure_policy(provider: _PrivacyPolicy) -> None: registry = PlatformRegistry() registry.register( ModuleManifest( id="policy_test", name="Policy test", version="test", capability_factories={ CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: lambda context: provider, }, ) ) registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) configure_runtime(registry=registry) def test_secure_default_returns_own_row_and_aggregate_counts(self) -> None: request = self._request() with patch( "govoplan_scheduling.backend.service.notification_dispatch_provider", return_value=object(), ): payload = self._participant_projection(request) response = SchedulingRequestResponse.model_validate(payload) self.assertEqual(request.participant_visibility, "aggregates_only") self.assertEqual(response.participant_visibility, "aggregates_only") self.assertEqual(response.effective_participant_visibility, "aggregates_only") self.assertEqual([participant.display_name for participant in response.participants], ["Alice"]) own = response.participants[0] self.assertTrue(own.is_current_participant) self.assertEqual(own.email, "alice@example.test") self.assertIsNone(own.respondent_id) self.assertIsNone(own.participant_type) self.assertIsNone(own.required) self.assertIsNone(own.poll_invitation_id) self.assertEqual(own.metadata, {}) self.assertEqual(response.participant_aggregate.total, 2) self.assertEqual(response.participant_aggregate.status_counts["responded"], 1) self.assertEqual(response.participant_aggregate.status_counts["invited"], 1) self.assertIsNone(response.tenant_id) self.assertIsNone(response.poll_id) self.assertIsNone(response.organizer_user_id) self.assertIsNone(response.calendar_integration_enabled) self.assertIsNone(response.calendar_id) self.assertIsNone(response.calendar_freebusy_enabled) self.assertIsNone(response.calendar_hold_enabled) self.assertIsNone(response.create_calendar_event_on_decision) self.assertIsNone(response.calendar_event_id) self.assertIsNone(response.public_participation_policy_enforcement_available) self.assertIsNone(response.participant_invitation_delivery_available) self.assertEqual(response.metadata, {}) slot = response.slots[0] self.assertIsNone(slot.poll_option_id) self.assertEqual(slot.freebusy_status, "busy") self.assertEqual( slot.freebusy_conflicts, [ { "start_at": "2026-07-27T09:30:00+00:00", "end_at": "2026-07-27T09:45:00+00:00", "status": "CONFIRMED", } ], ) self.assertIsNone(slot.tentative_hold_event_id) self.assertEqual(slot.metadata, {}) def test_configured_roster_returns_other_names_and_statuses_with_sensitive_fields_redacted(self) -> None: request = self._request(participant_visibility="names_and_statuses") response = SchedulingRequestResponse.model_validate(self._participant_projection(request)) self.assertEqual(response.effective_participant_visibility, "names_and_statuses") own = next(participant for participant in response.participants if participant.display_name == "Alice") other = next(participant for participant in response.participants if participant.display_name == "Bob") self.assertTrue(own.is_current_participant) self.assertIsNone(own.respondent_id) self.assertEqual(own.email, "alice@example.test") self.assertEqual(other.status, "invited") self.assertFalse(other.is_current_participant) self.assertIsNone(other.respondent_id) self.assertIsNone(other.email) self.assertIsNone(other.participant_type) self.assertIsNone(other.required) self.assertIsNone(other.poll_invitation_id) self.assertEqual(other.metadata, {}) def test_manager_and_organizer_bypass_participant_roster_restriction(self) -> None: request = self._request() manager = SchedulingRequestResponse.model_validate( scheduling_request_response( request, actor_ids=("manager-1",), actor_user_id="manager-1", can_manage=True, ) ) organizer = SchedulingRequestResponse.model_validate( scheduling_request_response( request, actor_ids=("organizer-1",), actor_user_id="organizer-1", ) ) for response in (manager, organizer): self.assertEqual(response.effective_participant_visibility, "names_and_statuses") self.assertEqual({participant.email for participant in response.participants}, { "alice@example.test", "bob@example.test", }) self.assertTrue(response.participant_visibility_decision.details["management_access"]) self.assertEqual(response.tenant_id, "tenant-1") self.assertEqual(response.poll_id, "poll-internal") self.assertEqual(response.organizer_user_id, "organizer-1") self.assertEqual(response.calendar_id, "calendar-internal") self.assertEqual(response.calendar_event_id, "event-internal") self.assertEqual(response.metadata["connector_secret_ref"], "secret-internal") self.assertEqual(response.slots[0].poll_option_id, "poll-option-internal") self.assertEqual( response.slots[0].freebusy_conflicts[0]["uid"], "connector-uid-internal", ) self.assertEqual(response.slots[0].tentative_hold_event_id, "hold-internal") self.assertFalse(response.participant_invitation_delivery_available) with patch( "govoplan_scheduling.backend.service.notification_dispatch_provider", return_value=object(), ): delivery_enabled = SchedulingRequestResponse.model_validate( scheduling_request_response( request, actor_ids=("manager-1",), actor_user_id="manager-1", can_manage=True, ) ) self.assertTrue(delivery_enabled.participant_invitation_delivery_available) def test_optional_policy_can_reduce_but_cannot_broaden_visibility(self) -> None: restricting_policy = _PrivacyPolicy("aggregates_only") self._configure_policy(restricting_policy) visible_request = self._request(participant_visibility="names_and_statuses") reduced = SchedulingRequestResponse.model_validate(self._participant_projection(visible_request)) self.assertEqual(reduced.effective_participant_visibility, "aggregates_only") self.assertEqual([participant.display_name for participant in reduced.participants], ["Alice"]) self.assertTrue(reduced.participant_visibility_decision.policy_applied) self.assertEqual(reduced.participant_visibility_decision.reason, "Tenant participant privacy policy") self.assertEqual(reduced.participant_visibility_decision.source_path, []) self.assertEqual(reduced.participant_visibility_decision.details, {}) self.assertEqual(restricting_policy.requests[0].actor_user_id, "alice-account") widening_policy = _PrivacyPolicy("names_and_statuses") self._configure_policy(widening_policy) private_request = self._request() clamped = SchedulingRequestResponse.model_validate(self._participant_projection(private_request)) self.assertEqual(clamped.effective_participant_visibility, "aggregates_only") self.assertEqual([participant.display_name for participant in clamped.participants], ["Alice"]) self.assertEqual(widening_policy.requests[0].requested_visibility, "aggregates_only") def test_create_and_update_schemas_expose_the_configuration(self) -> None: slot = SchedulingCandidateSlotInput.model_validate( { "start_at": "2026-07-20T09:00:00Z", "end_at": "2026-07-20T10:00:00Z", } ) created = SchedulingRequestCreateRequest(title="Privacy", slots=[slot]) updated = SchedulingRequestUpdateRequest(participant_visibility="names_and_statuses") self.assertEqual(created.participant_visibility, "aggregates_only") self.assertEqual(updated.participant_visibility, "names_and_statuses") if __name__ == "__main__": unittest.main()