feat: implement formal decision registry

2026-08-01 17:48:29 +02:00
commit ba8e42fd03
24 changed files with 1236 additions and 0 deletions
@@ -0,0 +1,101 @@
+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_decisions.backend.db.models import FormalDecisionRevision
+from govoplan_decisions.backend.service import DecisionStoreError, SqlDecisionRegistry
+
+
+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)
+
+
+if __name__ == "__main__":
+    unittest.main()