133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.institutional import (
|
|
InstitutionalReference,
|
|
ServiceAvailabilityRequirement,
|
|
ServiceDefinition,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_core.core.temporal import (
|
|
TemporalDataContext,
|
|
bind_temporal_data_context,
|
|
reset_temporal_data_context,
|
|
)
|
|
from govoplan_services.backend.db.models import ServiceDefinitionRevision
|
|
from govoplan_services.backend.service import (
|
|
RegistryServiceAvailabilityEvaluator,
|
|
ServiceStoreError,
|
|
SqlServiceDefinitionProvider,
|
|
record_service_definition,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str = "tenant-1"
|
|
account_id: str = "account-1"
|
|
|
|
|
|
class Registry:
|
|
def has(self, module_id: str) -> bool:
|
|
return module_id == "cases"
|
|
|
|
def has_capability(self, capability: str) -> bool:
|
|
return capability == "workflow.runtime"
|
|
|
|
|
|
def service(*, revision: str = "1", state: str = "published") -> ServiceDefinition:
|
|
recorded_at = NOW + timedelta(minutes=int(revision) - 1)
|
|
return ServiceDefinition(
|
|
reference=InstitutionalReference(kind="service", owner_module="services", object_id="permit", tenant_id="tenant-1", version=revision, valid_at=NOW),
|
|
key="permit",
|
|
temporal=TemporalRevision(revision=revision, valid_from=NOW, recorded_at=recorded_at, change_reason="Initial service." if revision == "1" else "Publication changed."),
|
|
title="Apply for a permit",
|
|
audience=("resident",),
|
|
channels=("portal",),
|
|
availability_requirements=(
|
|
ServiceAvailabilityRequirement(kind="module", reference="cases"),
|
|
ServiceAvailabilityRequirement(kind="capability", reference="workflow.runtime"),
|
|
),
|
|
publication_state=state, # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
class ServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
ServiceDefinitionRevision.__table__.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
self.principal = Principal()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_provider_returns_exact_published_revision(self) -> None:
|
|
first = record_service_definition(self.session, self.principal, definition=service())
|
|
provider = SqlServiceDefinitionProvider()
|
|
listed = provider.list_service_definitions(self.session, self.principal, tenant_id="tenant-1")
|
|
self.assertEqual((first,), tuple(listed))
|
|
exact = provider.get_service_definition(self.session, self.principal, reference=first.reference, effective_at=NOW)
|
|
self.assertEqual("1", exact.temporal.revision if exact else None)
|
|
|
|
suspended = service(revision="2", state="suspended")
|
|
record_service_definition(self.session, self.principal, definition=suspended, expected_revision="1")
|
|
historical = provider.get_service_definition(self.session, self.principal, reference=first.reference)
|
|
self.assertEqual("1", historical.temporal.revision if historical else None)
|
|
|
|
def test_occ_availability_and_tenant_boundary(self) -> None:
|
|
record_service_definition(self.session, self.principal, definition=service())
|
|
with self.assertRaisesRegex(ServiceStoreError, "stale"):
|
|
record_service_definition(self.session, self.principal, definition=service(revision="2"), expected_revision="0")
|
|
assessment = RegistryServiceAvailabilityEvaluator(Registry()).evaluate_service_availability(
|
|
self.session,
|
|
self.principal,
|
|
definition=service(),
|
|
effective_at=NOW,
|
|
)
|
|
self.assertEqual(
|
|
{"module:cases": True, "capability:workflow.runtime": True},
|
|
assessment.requirement_states,
|
|
)
|
|
with self.assertRaisesRegex(Exception, "cross tenants"):
|
|
SqlServiceDefinitionProvider().list_service_definitions(self.session, Principal("tenant-2"), tenant_id="tenant-1")
|
|
|
|
def test_read_context_can_reconstruct_recorded_state(self) -> None:
|
|
record_service_definition(self.session, self.principal, definition=service())
|
|
record_service_definition(
|
|
self.session,
|
|
self.principal,
|
|
definition=service(revision="2", state="suspended"),
|
|
expected_revision="1",
|
|
)
|
|
token = bind_temporal_data_context(
|
|
TemporalDataContext(
|
|
validity_mode="at",
|
|
valid_at=NOW + timedelta(hours=1),
|
|
recorded_at=NOW + timedelta(seconds=30),
|
|
)
|
|
)
|
|
try:
|
|
result = SqlServiceDefinitionProvider().list_service_definitions(
|
|
self.session,
|
|
self.principal,
|
|
tenant_id="tenant-1",
|
|
)
|
|
self.assertEqual("1", result[0].temporal.revision if result else None)
|
|
finally:
|
|
reset_temporal_data_context(token)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|