feat(policy): resolve scheduling participant privacy

tests/test_scheduling_privacy.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.admin.models import SystemSettings
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.policy import (
+    CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
+    SchedulingParticipantPrivacyPolicy,
+    SchedulingParticipantPrivacyRequest,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.tenancy.scope import Tenant, scope_registry
+from govoplan_policy.backend.manifest import manifest
+from govoplan_policy.backend.scheduling_privacy import (
+    SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY,
+    SqlSchedulingParticipantPrivacyPolicy,
+)
+
+
+def _policy_settings(maximum_visibility: object) -> dict[str, object]:
+    return {
+        SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: {
+            "maximum_visibility": maximum_visibility,
+        }
+    }
+
+
+class SchedulingParticipantPrivacyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(bind=self.engine, tables=[SystemSettings.__table__])
+        scope_registry.metadata.create_all(bind=self.engine, tables=[Tenant.__table__])
+        self.session = Session(self.engine)
+        self.provider = SqlSchedulingParticipantPrivacyPolicy()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _add_settings(
+        self,
+        *,
+        system: dict[str, object] | None = None,
+        tenant: dict[str, object] | None = None,
+    ) -> None:
+        self.session.add(SystemSettings(id="global", settings=system or {}))
+        self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings=tenant or {}))
+        self.session.flush()
+
+    def _resolve(self, requested_visibility: str = "names_and_statuses"):
+        return self.provider.resolve_scheduling_participant_visibility(
+            self.session,
+            request=SchedulingParticipantPrivacyRequest(
+                tenant_id="tenant-1",
+                scheduling_request_id="request-1",
+                participant_id="participant-1",
+                requested_visibility=requested_visibility,  # type: ignore[arg-type]
+                actor_user_id="user-1",
+            ),
+        )
+
+    def test_manifest_registers_the_core_capability(self) -> None:
+        factory = manifest.capability_factories[CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY]
+
+        provider = factory(ModuleContext(registry=object(), settings=object()))
+
+        self.assertIsInstance(provider, SchedulingParticipantPrivacyPolicy)
+
+    def test_missing_policy_preserves_scheduling_visibility(self) -> None:
+        self._add_settings()
+
+        decision = self._resolve()
+
+        self.assertEqual("names_and_statuses", decision.effective_visibility)
+        self.assertIsNone(decision.reason)
+        self.assertEqual((), decision.source_path)
+        self.assertEqual([], decision.details["configured_scopes"])
+
+    def test_missing_system_settings_is_an_unrestricted_read_only_default(self) -> None:
+        self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings={}))
+        self.session.flush()
+
+        decision = self._resolve()
+
+        self.assertEqual("names_and_statuses", decision.effective_visibility)
+        self.assertIsNone(self.session.get(SystemSettings, "global"))
+
+    def test_system_ceiling_restricts_participant_roster(self) -> None:
+        self._add_settings(system=_policy_settings("aggregates_only"))
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual("Participant roster visibility is restricted by policy.", decision.reason)
+        self.assertEqual(["system"], [step.path for step in decision.source_path])
+        self.assertEqual("aggregates_only", decision.source_path[0].policy["maximum_visibility"])
+
+    def test_tenant_ceiling_can_narrow_system_but_cannot_widen_it(self) -> None:
+        self._add_settings(
+            system=_policy_settings("aggregates_only"),
+            tenant=_policy_settings("names_and_statuses"),
+        )
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(["system", "tenant:tenant-1"], [step.path for step in decision.source_path])
+        self.assertEqual("aggregates_only", decision.details["policy_ceiling"])
+
+    def test_policy_never_broadens_an_aggregate_only_request(self) -> None:
+        self._add_settings(
+            system=_policy_settings("names_and_statuses"),
+            tenant=_policy_settings("names_and_statuses"),
+        )
+
+        decision = self._resolve("aggregates_only")
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertIsNone(decision.reason)
+
+    def test_invalid_explicit_policy_fails_closed_without_echoing_value(self) -> None:
+        self._add_settings(tenant=_policy_settings("surprise"))
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertIn("could not be validated", decision.reason or "")
+        self.assertEqual(
+            [{"scope": "tenant", "code": "invalid_maximum_visibility"}],
+            decision.details["configuration_errors"],
+        )
+        self.assertNotIn("surprise", repr(decision.to_dict()))
+        self.assertEqual("invalid_fail_closed", decision.source_path[0].policy["configuration_status"])
+
+    def test_invalid_policy_shape_fails_closed(self) -> None:
+        self._add_settings(
+            system={SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: "not-an-object"},
+        )
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(
+            [{"scope": "system", "code": "invalid_policy_shape"}],
+            decision.details["configuration_errors"],
+        )
+
+    def test_missing_tenant_fails_closed(self) -> None:
+        self.session.add(SystemSettings(id="global", settings={}))
+        self.session.flush()
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(
+            [{"scope": "tenant", "code": "tenant_not_found"}],
+            decision.details["configuration_errors"],
+        )
+        self.assertEqual(["tenant:tenant-1"], [step.path for step in decision.source_path])
+
+    def test_invalid_requested_visibility_fails_closed(self) -> None:
+        self._add_settings()
+
+        decision = self._resolve("invalid")
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(
+            [{"scope": "request", "code": "invalid_requested_visibility"}],
+            decision.details["configuration_errors"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()