feat(scheduling): use central people picker

tests/test_people_search.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+
+from fastapi import HTTPException
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.people import (
+    CAPABILITY_ACCESS_PEOPLE_SEARCH,
+    PeopleSearchGroup,
+    PersonSearchCandidate,
+)
+from govoplan_scheduling.backend.manifest import WRITE_SCOPE
+from govoplan_scheduling.backend.router import api_search_scheduling_people
+from govoplan_scheduling.backend.runtime import configure_runtime
+
+
+class _Registry:
+    def __init__(self, capabilities: dict[str, object] | None = None) -> None:
+        self.capabilities = capabilities or {}
+
+    def has_capability(self, name: str) -> bool:
+        return name in self.capabilities
+
+    def capability(self, name: str) -> object:
+        return self.capabilities[name]
+
+
+class _PeopleProvider:
+    def __init__(self) -> None:
+        self.calls: list[tuple[object, object, str, int]] = []
+
+    def search_people(
+        self,
+        session: object,
+        principal: object,
+        *,
+        query: str,
+        limit: int = 25,
+    ) -> tuple[PeopleSearchGroup, ...]:
+        self.calls.append((session, principal, query, limit))
+        return (
+            PeopleSearchGroup(
+                key="accounts",
+                label="Accounts",
+                candidates=(
+                    PersonSearchCandidate(
+                        selection_key="account:account-2",
+                        kind="account",
+                        reference_id="account-2",
+                        display_name="Ada Lovelace",
+                        email="ada@example.test",
+                        source_module="access",
+                        source_label="Accounts",
+                        source_ref="access:account:account-2",
+                        source_revision="revision-1",
+                        description="Research",
+                        provenance={"tenant_id": "tenant-1", "internal": "secret"},
+                        metadata={"internal_group_ids": ["group-1"]},
+                    ),
+                ),
+            ),
+        )
+
+
+def _principal(*scopes: str) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id="tenant-1",
+            email="organizer@example.test",
+            display_name="Organizer",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="membership-1"),
+    )
+
+
+class SchedulingPeopleSearchTests(unittest.TestCase):
+    def tearDown(self) -> None:
+        configure_runtime(registry=_Registry())
+
+    def test_search_uses_principal_aware_core_aggregator_and_redacts_provider_internals(self) -> None:
+        provider = _PeopleProvider()
+        registry = _Registry({CAPABILITY_ACCESS_PEOPLE_SEARCH: provider})
+        configure_runtime(registry=registry)
+        session = object()
+        principal = _principal(WRITE_SCOPE)
+
+        response = api_search_scheduling_people(
+            query="ada",
+            limit=12,
+            session=session,  # type: ignore[arg-type] - provider contract is intentionally generic
+            principal=principal,
+        )
+
+        self.assertEqual([(session, principal, "ada", 12)], provider.calls)
+        payload = response.model_dump()
+        self.assertEqual("account:account-2", payload["groups"][0]["candidates"][0]["selection_key"])
+        self.assertEqual("revision-1", payload["groups"][0]["candidates"][0]["source_revision"])
+        self.assertNotIn("source_ref", payload["groups"][0]["candidates"][0])
+        self.assertNotIn("provenance", payload["groups"][0]["candidates"][0])
+        self.assertNotIn("metadata", payload["groups"][0]["candidates"][0])
+
+    def test_search_is_empty_when_no_optional_directory_provider_is_installed(self) -> None:
+        configure_runtime(registry=_Registry())
+
+        response = api_search_scheduling_people(
+            query="ada",
+            limit=25,
+            session=object(),  # type: ignore[arg-type]
+            principal=_principal(WRITE_SCOPE),
+        )
+
+        self.assertEqual([], response.groups)
+
+    def test_search_requires_scheduling_write_or_admin_access(self) -> None:
+        provider = _PeopleProvider()
+        configure_runtime(registry=_Registry({CAPABILITY_ACCESS_PEOPLE_SEARCH: provider}))
+
+        with self.assertRaises(HTTPException) as raised:
+            api_search_scheduling_people(
+                query="ada",
+                limit=25,
+                session=object(),  # type: ignore[arg-type]
+                principal=_principal("scheduling:schedule:read"),
+            )
+
+        self.assertEqual(403, raised.exception.status_code)
+        self.assertEqual([], provider.calls)
+
+
+if __name__ == "__main__":
+    unittest.main()