feat(policy): add scheduling participant privacy contract

This commit is contained in:
2026-07-20 18:23:59 +02:00
parent 2f559e3f0b
commit abbef5a10b
2 changed files with 159 additions and 0 deletions

View File

@@ -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

View File

@@ -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()