272 lines
8.7 KiB
Python
272 lines
8.7 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_core.core.records import RecordContractError, RecordSourceLocator
|
|
from govoplan_decisions.backend.db.models import FormalDecisionRevision
|
|
from govoplan_decisions.backend.record_source import DecisionsRecordSource
|
|
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"
|
|
scopes: tuple[str, ...] = (
|
|
"decisions:decision:read",
|
|
"decisions:decision:read_sensitive",
|
|
)
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope in self.scopes
|
|
|
|
|
|
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)
|
|
|
|
def test_record_source_requires_protected_read_and_hashes_exact_revision(
|
|
self,
|
|
) -> None:
|
|
SqlDecisionRegistry().record_decision(
|
|
self.session,
|
|
self.principal,
|
|
decision=decision(),
|
|
)
|
|
locator = RecordSourceLocator(
|
|
tenant_id="tenant-1",
|
|
source_module="decisions",
|
|
resource_type="decision_revision",
|
|
resource_id="decision-1",
|
|
source_revision="1",
|
|
)
|
|
result = DecisionsRecordSource().resolve(
|
|
self.session,
|
|
self.principal,
|
|
locator=locator,
|
|
purpose="file formal outcome",
|
|
)
|
|
self.assertEqual(64, len(result.content_sha256 or ""))
|
|
self.assertEqual("decided", result.metadata["state"])
|
|
self.assertTrue(result.metadata["protected_snapshot"])
|
|
|
|
with self.assertRaisesRegex(RecordContractError, "Protected Decision"):
|
|
DecisionsRecordSource().resolve(
|
|
self.session,
|
|
Principal(scopes=("decisions:decision:read",)),
|
|
locator=locator,
|
|
purpose="file formal outcome",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|