175 lines
5.3 KiB
Python
175 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
|
|
from govoplan_core.core.institutional import (
|
|
GovernedContextEnvelope,
|
|
InstitutionalReference,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_core.core.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
SearchResourceReference,
|
|
)
|
|
from govoplan_cases.backend.db.models import (
|
|
CaseAccessGrant,
|
|
CaseIdentity,
|
|
CaseRecordRevision,
|
|
CaseStatusDefinition,
|
|
CaseTimelineEntry,
|
|
CaseTypeDefinition,
|
|
)
|
|
from govoplan_cases.backend.domain import CaseRecord
|
|
from govoplan_cases.backend.search_source import CasesSearchSource, PROVIDER_ID
|
|
from govoplan_cases.backend.service import (
|
|
create_case,
|
|
upsert_case_status,
|
|
upsert_case_type,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 6, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
class CasesSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
for table in (
|
|
CaseStatusDefinition.__table__,
|
|
CaseTypeDefinition.__table__,
|
|
CaseIdentity.__table__,
|
|
CaseRecordRevision.__table__,
|
|
CaseAccessGrant.__table__,
|
|
CaseTimelineEntry.__table__,
|
|
):
|
|
table.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
principal = _principal({"cases:case:read", "cases:case:create"})
|
|
upsert_case_status(
|
|
self.session,
|
|
principal,
|
|
status_key="review",
|
|
label="Review",
|
|
)
|
|
upsert_case_type(
|
|
self.session,
|
|
principal,
|
|
type_key="permit",
|
|
label="Permit",
|
|
initial_status_key="review",
|
|
)
|
|
case_ref = _ref("case", "case-1", "cases")
|
|
create_case(
|
|
self.session,
|
|
principal,
|
|
record=CaseRecord(
|
|
reference=case_ref,
|
|
case_number="PERMIT-1",
|
|
case_type_key="permit",
|
|
status_key="review",
|
|
title="Permit application",
|
|
context=GovernedContextEnvelope(
|
|
tenant_id="tenant-1",
|
|
temporal=TemporalRevision(
|
|
revision="1",
|
|
valid_from=NOW,
|
|
recorded_at=NOW,
|
|
change_reason="Created.",
|
|
),
|
|
case_ref=case_ref,
|
|
),
|
|
opened_at=NOW,
|
|
recorded_at=NOW,
|
|
change_reason="Created.",
|
|
),
|
|
idempotency_key="case-create",
|
|
)
|
|
self.session.commit()
|
|
self.source = CasesSearchSource()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_backfill_authorization_and_event_change_are_bounded(self) -> None:
|
|
page = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type="case",
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
self.assertEqual(("case-1",), tuple(item.resource_id for item in page.documents))
|
|
self.assertEqual("PERMIT-1", page.documents[0].metadata["case_number"])
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({"cases:case:read"}),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal(set()),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
changes = self.source.index_changes_for_event(
|
|
self.session,
|
|
event=PlatformEvent(
|
|
type="cases.case.updated",
|
|
module_id="cases",
|
|
tenant=EventTenantRef(id="tenant-1"),
|
|
resource=EventObjectRef(type="case", id="case-1"),
|
|
),
|
|
delivery_key="delivery-1",
|
|
)
|
|
self.assertEqual("upsert", changes[0].kind)
|
|
|
|
|
|
def _ref(kind: str, object_id: str, owner: str) -> InstitutionalReference:
|
|
return InstitutionalReference(
|
|
kind=kind, # type: ignore[arg-type]
|
|
owner_module=owner,
|
|
object_id=object_id,
|
|
tenant_id="tenant-1",
|
|
version="1",
|
|
valid_at=NOW,
|
|
)
|
|
|
|
|
|
def _principal(scopes: set[str]) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="membership-1"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|