132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import Account, User
|
|
from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent
|
|
from govoplan_calendar.backend.search_source import (
|
|
CalendarSearchSource,
|
|
PROVIDER_ID,
|
|
RESOURCE_TYPE,
|
|
)
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
SearchResourceReference,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
class CalendarSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(
|
|
Account.__table__,
|
|
User.__table__,
|
|
CalendarCollection.__table__,
|
|
CalendarEvent.__table__,
|
|
),
|
|
)
|
|
self.session = Session(self.engine)
|
|
start = datetime(2026, 8, 5, 9, tzinfo=timezone.utc)
|
|
self.session.add_all(
|
|
(
|
|
Account(
|
|
id="account-1",
|
|
email="one@example.test",
|
|
normalized_email="one@example.test",
|
|
),
|
|
User(
|
|
id="user-1",
|
|
tenant_id="tenant-1",
|
|
account_id="account-1",
|
|
email="one@example.test",
|
|
),
|
|
CalendarCollection(
|
|
id="calendar-1",
|
|
tenant_id="tenant-1",
|
|
slug="private",
|
|
name="Private calendar",
|
|
owner_type="user",
|
|
owner_id="user-1",
|
|
visibility="private",
|
|
created_by_user_id="user-1",
|
|
),
|
|
CalendarEvent(
|
|
id="event-1",
|
|
tenant_id="tenant-1",
|
|
calendar_id="calendar-1",
|
|
uid="event-1@example.test",
|
|
summary="Permit review",
|
|
description="Review monthly permits",
|
|
start_at=start,
|
|
end_at=start + timedelta(hours=1),
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
self.source = CalendarSearchSource()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_private_calendar_is_visible_only_to_current_owner(self) -> None:
|
|
page = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type=RESOURCE_TYPE,
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
self.assertEqual(("event-1",), tuple(doc.resource_id for doc in page.documents))
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="calendar",
|
|
resource_type=RESOURCE_TYPE,
|
|
resource_id="event-1",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal("user-1"),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal("user-2"),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
|
|
|
|
def _principal(user_id: str) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=f"account-{user_id}",
|
|
membership_id=user_id,
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset({"calendar:event:read"}),
|
|
),
|
|
account=SimpleNamespace(id=f"account-{user_id}"),
|
|
user=SimpleNamespace(id=user_id),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|