130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
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,
|
|
MandateDefinition,
|
|
MandateResolutionRequest,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_mandates.backend.db.models import MandateRevision
|
|
from govoplan_mandates.backend.service import (
|
|
MandateStoreError,
|
|
SqlMandateResolver,
|
|
get_mandate,
|
|
record_mandate,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 9, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str = "tenant-1"
|
|
account_id: str = "account-1"
|
|
|
|
|
|
def definition(*, revision: str = "1", status: str = "active") -> MandateDefinition:
|
|
temporal = TemporalRevision(
|
|
revision=revision,
|
|
valid_from=NOW,
|
|
recorded_at=NOW + timedelta(minutes=int(revision) - 1),
|
|
change_reason="Initial authority." if revision == "1" else "Authority changed.",
|
|
)
|
|
return MandateDefinition(
|
|
reference=InstitutionalReference(
|
|
kind="mandate",
|
|
owner_module="mandates",
|
|
object_id="committee-permit",
|
|
tenant_id="tenant-1",
|
|
version=revision,
|
|
valid_at=NOW,
|
|
),
|
|
temporal=temporal,
|
|
task_types=("committee.formal_decision",),
|
|
authority_types=("permit",),
|
|
status=status, # type: ignore[arg-type]
|
|
suspension_reason="Review pending." if status == "suspended" else None,
|
|
)
|
|
|
|
|
|
class MandateTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
MandateRevision.__table__.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
self.principal = Principal()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_record_resolve_replay_and_occ(self) -> None:
|
|
first = definition()
|
|
recorded = record_mandate(self.session, self.principal, definition=first)
|
|
self.assertEqual("1", recorded.temporal.revision)
|
|
replay = record_mandate(self.session, self.principal, definition=first)
|
|
self.assertEqual("1", replay.temporal.revision)
|
|
|
|
request = MandateResolutionRequest(
|
|
tenant_id="tenant-1",
|
|
effective_at=NOW + timedelta(hours=1),
|
|
task_type="committee.formal_decision",
|
|
authority_type="permit",
|
|
)
|
|
resolved = SqlMandateResolver().resolve_mandate(
|
|
self.session,
|
|
self.principal,
|
|
request=request,
|
|
)
|
|
self.assertTrue(resolved.competent)
|
|
self.assertEqual("1", resolved.mandates[0].temporal.revision)
|
|
|
|
suspended = definition(revision="2", status="suspended")
|
|
record_mandate(
|
|
self.session,
|
|
self.principal,
|
|
definition=suspended,
|
|
expected_revision="1",
|
|
)
|
|
self.assertFalse(
|
|
SqlMandateResolver().resolve_mandate(
|
|
self.session,
|
|
self.principal,
|
|
request=request,
|
|
).competent
|
|
)
|
|
with self.assertRaisesRegex(MandateStoreError, "stale"):
|
|
stale = replace(
|
|
suspended,
|
|
reference=replace(suspended.reference, version="3"),
|
|
temporal=replace(suspended.temporal, revision="3"),
|
|
)
|
|
record_mandate(
|
|
self.session,
|
|
self.principal,
|
|
definition=stale,
|
|
expected_revision="1",
|
|
)
|
|
|
|
def test_tenant_isolation(self) -> None:
|
|
record_mandate(self.session, self.principal, definition=definition())
|
|
self.assertIsNone(
|
|
get_mandate(
|
|
self.session,
|
|
Principal(tenant_id="tenant-2"),
|
|
mandate_id="committee-permit",
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|