245 lines
10 KiB
Python
245 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
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",
|
|
"organizer_user_id": "organizer-1",
|
|
}
|
|
if participant_visibility is not None:
|
|
values["participant_visibility"] = participant_visibility
|
|
request = SchedulingRequest(**values)
|
|
request.participants = [
|
|
SchedulingParticipant(
|
|
tenant_id="tenant-1",
|
|
respondent_id="alice-id",
|
|
display_name="Alice",
|
|
email="alice@example.test",
|
|
participant_type="internal",
|
|
status="responded",
|
|
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()
|
|
|
|
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"])
|
|
self.assertEqual(response.participants[0].email, "alice@example.test")
|
|
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)
|
|
|
|
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.assertEqual(own.respondent_id, "alice-id")
|
|
self.assertEqual(own.email, "alice@example.test")
|
|
self.assertEqual(other.status, "invited")
|
|
self.assertIsNone(other.respondent_id)
|
|
self.assertIsNone(other.email)
|
|
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"])
|
|
|
|
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[0]["path"], "tenant:tenant-1")
|
|
self.assertTrue(reduced.participant_visibility_decision.details["provider_session_available"])
|
|
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()
|