From abbef5a10b7130c6626310e7d0b585bdda609613 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 20 Jul 2026 18:23:59 +0200 Subject: [PATCH] feat(policy): add scheduling participant privacy contract --- src/govoplan_core/core/policy.py | 64 +++++++++++++++ tests/test_scheduling_privacy_contract.py | 95 +++++++++++++++++++++++ 2 files changed, 159 insertions(+) create mode 100644 tests/test_scheduling_privacy_contract.py diff --git a/src/govoplan_core/core/policy.py b/src/govoplan_core/core/policy.py index a76d8d5..a5512b2 100644 --- a/src/govoplan_core/core/policy.py +++ b/src/govoplan_core/core/policy.py @@ -5,8 +5,10 @@ from typing import Any, Iterable, Literal, Mapping, Protocol, cast, runtime_chec from urllib.parse import quote, unquote PolicyScopeType = Literal["system", "tenant", "user", "group", "campaign"] +SchedulingParticipantVisibility = Literal["aggregates_only", "names_and_statuses"] CAPABILITY_POLICY_PRIVACY_RETENTION = "policy.privacyRetention" +CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY = "policy.schedulingParticipantPrivacy" POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ("system", "tenant", "user", "group", "campaign") @@ -126,6 +128,55 @@ class PolicyDecision: } +@dataclass(frozen=True, slots=True) +class SchedulingParticipantPrivacyRequest: + """Context for resolving what one Scheduling participant may see. + + ``requested_visibility`` is the Scheduling-owned configuration result. A + policy provider may preserve or reduce it, but must not broaden it. Core + deliberately defines no fallback when the optional capability is absent; + that secure default remains the responsibility of Scheduling. + """ + + tenant_id: str + scheduling_request_id: str + participant_id: str + requested_visibility: SchedulingParticipantVisibility + actor_user_id: str | None = None + context: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class SchedulingParticipantPrivacyDecision: + """Policy-resolved visibility of other participants' names and statuses.""" + + effective_visibility: SchedulingParticipantVisibility + reason: str | None = None + source_path: tuple[PolicySourceStep, ...] = () + details: Mapping[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "effective_visibility": self.effective_visibility, + "reason": self.reason, + "source_path": [step.to_dict() for step in self.source_path], + "details": dict(self.details), + } + + +@runtime_checkable +class SchedulingParticipantPrivacyPolicy(Protocol): + """Optional policy hook for Scheduling participant-roster disclosure.""" + + def resolve_scheduling_participant_visibility( + self, + session: object, + *, + request: SchedulingParticipantPrivacyRequest, + ) -> SchedulingParticipantPrivacyDecision: + ... + + @runtime_checkable class PrivacyRetentionService(Protocol): def privacy_policy_from_settings(self, *args: Any, **kwargs: Any) -> Any: @@ -160,3 +211,16 @@ class PrivacyRetentionService(Protocol): def apply_retention_policy(self, *args: Any, **kwargs: Any) -> Any: ... + + +def scheduling_participant_privacy_policy( + registry: object | None, +) -> SchedulingParticipantPrivacyPolicy | None: + """Return the optional provider without selecting a visibility fallback.""" + + if registry is None or not hasattr(registry, "has_capability"): + return None + if not registry.has_capability(CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY): + return None + capability = registry.capability(CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY) + return capability if isinstance(capability, SchedulingParticipantPrivacyPolicy) else None diff --git a/tests/test_scheduling_privacy_contract.py b/tests/test_scheduling_privacy_contract.py new file mode 100644 index 0000000..945a1f2 --- /dev/null +++ b/tests/test_scheduling_privacy_contract.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, + PolicySourceStep, + SchedulingParticipantPrivacyDecision, + SchedulingParticipantPrivacyPolicy, + SchedulingParticipantPrivacyRequest, + scheduling_participant_privacy_policy, +) +from govoplan_core.core.registry import PlatformRegistry + + +class _PrivacyPolicy: + def resolve_scheduling_participant_visibility( + self, + session: object, + *, + request: SchedulingParticipantPrivacyRequest, + ) -> SchedulingParticipantPrivacyDecision: + del session + return SchedulingParticipantPrivacyDecision( + effective_visibility="aggregates_only", + reason=f"Participant roster is restricted for {request.scheduling_request_id}.", + details={"requested_visibility": request.requested_visibility}, + ) + + +class SchedulingPrivacyContractTests(unittest.TestCase): + def test_capability_name_and_request_context_are_stable(self) -> None: + self.assertEqual( + CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, + "policy.schedulingParticipantPrivacy", + ) + request = SchedulingParticipantPrivacyRequest( + tenant_id="tenant-1", + scheduling_request_id="request-1", + participant_id="participant-1", + actor_user_id="user-1", + requested_visibility="names_and_statuses", + context={"request_status": "open"}, + ) + + self.assertEqual(request.requested_visibility, "names_and_statuses") + self.assertEqual(request.context["request_status"], "open") + + def test_decision_serializes_policy_provenance(self) -> None: + decision = SchedulingParticipantPrivacyDecision( + effective_visibility="aggregates_only", + reason="Tenant privacy policy restricts the participant roster.", + source_path=( + PolicySourceStep( + scope_type="tenant", + scope_id="tenant-1", + label="Tenant", + applied_fields=("scheduling_participant_visibility",), + ), + ), + details={"requested_visibility": "names_and_statuses"}, + ) + + payload = decision.to_dict() + self.assertEqual(payload["effective_visibility"], "aggregates_only") + self.assertEqual(payload["source_path"][0]["path"], "tenant:tenant-1") + self.assertEqual(payload["details"]["requested_visibility"], "names_and_statuses") + + def test_optional_provider_resolves_through_platform_registry(self) -> None: + provider = _PrivacyPolicy() + self.assertIsInstance(provider, SchedulingParticipantPrivacyPolicy) + + 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())) + + self.assertIs(scheduling_participant_privacy_policy(registry), provider) + + def test_optional_provider_lookup_does_not_choose_a_fallback(self) -> None: + self.assertIsNone(scheduling_participant_privacy_policy(None)) + self.assertIsNone(scheduling_participant_privacy_policy(PlatformRegistry())) + + +if __name__ == "__main__": + unittest.main()