167 lines
6.6 KiB
Python
167 lines
6.6 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 (
|
|
ActorRepresentationReference,
|
|
EvidenceReference,
|
|
FormalDecision,
|
|
GovernedContextEnvelope,
|
|
InstitutionalReference,
|
|
LegalBasisReference,
|
|
TemporalRevision,
|
|
revise_formal_decision,
|
|
)
|
|
from govoplan_core.core.temporal import (
|
|
TemporalDataContext,
|
|
bind_temporal_data_context,
|
|
reset_temporal_data_context,
|
|
)
|
|
from govoplan_decisions.backend.db.models import FormalDecisionRevision
|
|
from govoplan_decisions.backend.service import (
|
|
DecisionStoreError,
|
|
SqlDecisionRegistry,
|
|
list_decisions,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str = "tenant-1"
|
|
account_id: str = "account-1"
|
|
|
|
|
|
def reference(kind: str, object_id: str, owner: str, version: str | None = "1") -> InstitutionalReference:
|
|
return InstitutionalReference(kind=kind, owner_module=owner, object_id=object_id, tenant_id="tenant-1", version=version, valid_at=NOW) # type: ignore[arg-type]
|
|
|
|
|
|
def decision() -> FormalDecision:
|
|
temporal = TemporalRevision(revision="1", valid_from=NOW, recorded_at=NOW, change_reason="Decision accepted.")
|
|
decision_ref = reference("decision", "decision-1", "decisions")
|
|
mandate_ref = reference("mandate", "mandate-1", "mandates", "7")
|
|
evidence = EvidenceReference(kind="record", owner_module="committee", evidence_id="minutes-1", tenant_id="tenant-1", version="1", captured_at=NOW)
|
|
basis = LegalBasisReference(kind="statute", authority="Council", reference="rules:12", version="2026")
|
|
context = GovernedContextEnvelope(
|
|
tenant_id="tenant-1",
|
|
temporal=temporal,
|
|
actor=ActorRepresentationReference(tenant_id="tenant-1", account_id="account-1", identity_id="identity-1", mandate_ref=mandate_ref),
|
|
mandate_ref=mandate_ref,
|
|
decision_ref=decision_ref,
|
|
legal_bases=(basis,),
|
|
evidence=(evidence,),
|
|
)
|
|
return FormalDecision(
|
|
reference=decision_ref,
|
|
temporal=temporal,
|
|
decision_type="committee.resolution",
|
|
subject_refs=(reference("case", "case-1", "cases"),),
|
|
state="decided",
|
|
authority_context=context,
|
|
fact_evidence=(evidence,),
|
|
legal_bases=(basis,),
|
|
operative_result="Approved.",
|
|
reasoning="Requirements are met.",
|
|
)
|
|
|
|
|
|
class DecisionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
FormalDecisionRevision.__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_replay_revise_and_exact_history(self) -> None:
|
|
registry = SqlDecisionRegistry()
|
|
first = registry.record_decision(self.session, self.principal, decision=decision())
|
|
self.assertEqual(first, registry.record_decision(self.session, self.principal, decision=decision()))
|
|
temporal = TemporalRevision(revision="2", valid_from=NOW, recorded_at=NOW + timedelta(minutes=1), change_reason="Effect confirmed.")
|
|
revised = revise_formal_decision(first, expected_revision="1", temporal=temporal, state="effective")
|
|
registry.record_decision(self.session, self.principal, decision=revised, expected_revision="1")
|
|
current = registry.get_decision(self.session, self.principal, reference=reference("decision", "decision-1", "decisions", None))
|
|
historical = registry.get_decision(self.session, self.principal, reference=first.reference)
|
|
self.assertEqual("2", current.temporal.revision if current else None)
|
|
self.assertEqual("1", historical.temporal.revision if historical else None)
|
|
|
|
def test_occ_and_tenant_isolation(self) -> None:
|
|
registry = SqlDecisionRegistry()
|
|
first = registry.record_decision(self.session, self.principal, decision=decision())
|
|
temporal = TemporalRevision(revision="2", valid_from=NOW, recorded_at=NOW + timedelta(minutes=1), change_reason="Effect confirmed.")
|
|
revised = revise_formal_decision(first, expected_revision="1", temporal=temporal, state="effective")
|
|
with self.assertRaisesRegex(DecisionStoreError, "stale"):
|
|
registry.record_decision(self.session, self.principal, decision=revised, expected_revision="0")
|
|
with self.assertRaisesRegex(Exception, "same-tenant"):
|
|
registry.get_decision(self.session, Principal("tenant-2"), reference=first.reference)
|
|
|
|
def test_temporal_context_selects_valid_and_recorded_state(self) -> None:
|
|
registry = SqlDecisionRegistry()
|
|
first = registry.record_decision(self.session, self.principal, decision=decision())
|
|
temporal = TemporalRevision(
|
|
revision="2",
|
|
valid_from=NOW,
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Effect confirmed.",
|
|
)
|
|
revised = revise_formal_decision(
|
|
first,
|
|
expected_revision="1",
|
|
temporal=temporal,
|
|
state="effective",
|
|
)
|
|
registry.record_decision(
|
|
self.session,
|
|
self.principal,
|
|
decision=revised,
|
|
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:
|
|
historical = registry.get_decision(
|
|
self.session,
|
|
self.principal,
|
|
reference=reference("decision", "decision-1", "decisions", None),
|
|
)
|
|
self.assertEqual("1", historical.temporal.revision if historical else None)
|
|
finally:
|
|
reset_temporal_data_context(token)
|
|
|
|
token = bind_temporal_data_context(
|
|
TemporalDataContext(
|
|
validity_mode="at",
|
|
valid_at=NOW - timedelta(days=1),
|
|
)
|
|
)
|
|
try:
|
|
self.assertEqual((), list_decisions(self.session, self.principal))
|
|
finally:
|
|
reset_temporal_data_context(token)
|
|
|
|
token = bind_temporal_data_context(TemporalDataContext(validity_mode="all"))
|
|
try:
|
|
self.assertEqual(1, len(list_decisions(self.session, self.principal)))
|
|
finally:
|
|
reset_temporal_data_context(token)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|