from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime, timedelta import json from pathlib import Path import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.records import ( RecordArchiveProviderState, RecordArchiveReceipt, RecordFilingRequest, RecordSourceLocator, RecordSourceReference, ) from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation from govoplan_core.core.approvals import ApprovalCheck, ApprovalRequestRef from govoplan_core.core.temporal import ( TemporalDataContext, bind_temporal_data_context, reset_temporal_data_context, ) from govoplan_records.backend.db.models import ( RecordChronologyEntry, RecordClassRevision, RecordDispositionRevision, RecordFilePlanRevision, RecordHoldRevision, RecordIdentity, RecordItem, RecordRevision, RecordTransferPackageRevision, RecordVolumeRevision, ) from govoplan_records.backend.archive import SimulatedRecordArchiveProvider from govoplan_records.backend.service import RecordConflictError, SqlRecordRegistry NOW = datetime(2026, 1, 6, 9, 0, tzinfo=UTC) @dataclass class Principal: tenant_id: str = "tenant-1" account_id: str = "account-1" acting_assignment_id: str = "assignment-1" acting_delegation_id: str | None = None class SourceProvider: provider_id = "files" def resource_types(self): return ("file_version",) def resolve(self, session, principal, *, locator, purpose): del session, purpose if principal.tenant_id != locator.tenant_id: raise ValueError("Source access denied.") return RecordSourceReference( locator=locator, label="Decision.pdf", authority_mode="external_authoritative", content_sha256="a" * 64, content_type="application/pdf", size_bytes=2048, valid_from=NOW, recorded_at=NOW, launch_url="/files?fileId=file-1", ) class UnknownOutcomeArchiveProvider: provider_id = "unknown_simulation" def __init__(self) -> None: self.dispatch_count = 0 def state(self): return RecordArchiveProviderState( provider_id=self.provider_id, label="Unknown-outcome simulation", profiles=("govoplan-unknown-outcome-v1",), authority_modes=("linked_reference",), healthy=True, checked_at=NOW, simulated=True, ) def dispatch(self, session, principal, *, request): del session, principal self.dispatch_count += 1 return RecordArchiveReceipt( provider_id=self.provider_id, package_id=request.package.package_id, outcome="outcome_unknown", observed_at=NOW + timedelta(minutes=12), receipt_sha256="b" * 64, retry_safe=False, simulated=True, metadata={"custody_transferred": False}, ) class Registry: def __init__(self) -> None: self.provider = SourceProvider() self.archive_provider = SimulatedRecordArchiveProvider() self.unknown_archive_provider = UnknownOutcomeArchiveProvider() self.approvals = ApprovalProvider() def capability_names(self): return ( "records.source.files", "records.archive.simulation", "records.archive.unknown_simulation", ) def tenant_capability(self, name, session, *, tenant_id): del session if tenant_id != "tenant-1": return None return { "records.source.files": self.provider, "records.archive.simulation": self.archive_provider, "records.archive.unknown_simulation": self.unknown_archive_provider, "approvals.requests": self.approvals, }.get(name) class ApprovalProvider: def __init__(self) -> None: self.approved = False self.request = None def create_request(self, session, principal, *, command, idempotency_key): del session, principal, idempotency_key self.request = command return ApprovalRequestRef(id="approval-1", revision=1, state="pending") def check_approved( self, session, principal, *, request_id, subject_module, subject_type, subject_id, subject_version, subject_digest, ): del session, principal return ApprovalCheck( request_id=request_id, revision=2 if self.approved else 1, state="approved" if self.approved else "pending", approved=self.approved, subject_module=subject_module, subject_type=subject_type, subject_id=subject_id, subject_version=subject_version, subject_digest=subject_digest, ) def create_template(self, *args, **kwargs): raise NotImplementedError def revise_template(self, *args, **kwargs): raise NotImplementedError def publish_template(self, *args, **kwargs): raise NotImplementedError def get_request(self, *args, **kwargs): return None def decide(self, *args, **kwargs): raise NotImplementedError class RecordsTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") for table in ( RecordFilePlanRevision.__table__, RecordClassRevision.__table__, RecordIdentity.__table__, RecordRevision.__table__, RecordVolumeRevision.__table__, RecordItem.__table__, RecordChronologyEntry.__table__, RecordHoldRevision.__table__, RecordDispositionRevision.__table__, RecordTransferPackageRevision.__table__, RecoveryOperation.__table__, RecoveryCheckpoint.__table__, ): table.create(self.engine) self.session = Session(self.engine) self.principal = Principal() self.records = SqlRecordRegistry(Registry()) self._seed_catalog() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed_catalog(self) -> None: self.records.write_file_plan_node( self.session, self.principal, payload={ "node_id": "plan-permits", "code": "10.20", "label": "Permits", "recorded_at": NOW, "valid_from": NOW, "idempotency_key": "catalog-node-1", "institutional_context": {"organization_unit_id": "unit-1"}, }, ) self.records.write_record_class( self.session, self.principal, payload={ "class_id": "class-permit", "file_plan_node_id": "plan-permits", "key": "permit.application", "label": "Permit application", "allowed_source_types": ["files:file_version"], "retention_period_days": 3650, "access_mode": "tenant", "recorded_at": NOW, "valid_from": NOW, "idempotency_key": "catalog-class-1", "institutional_context": {"organization_unit_id": "unit-1"}, }, ) self.session.commit() def _create_record(self) -> dict[str, object]: result = self.records.create_record( self.session, self.principal, payload={ "record_id": "record-1", "record_number": "2026/0001", "class_id": "class-permit", "file_plan_node_id": "plan-permits", "title": "Permit application Ada Example", "description": "Assisted service journey", "state": "open", "source_authority_mode": "native_authoritative", "access_mode": "tenant", "purpose": "process permit application", "institutional_context": { "organization_unit_id": "unit-1", "case_id": "case-1", }, "recorded_at": NOW + timedelta(minutes=1), "valid_from": NOW, "change_reason": "Application received.", "idempotency_key": "record-create-1", }, ) self.session.commit() return result def test_record_creation_revision_and_temporal_reconstruction(self) -> None: first = self._create_record() replay = self._create_record() self.assertEqual(first["record_id"], replay["record_id"]) revised = self.records.update_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 1, "title": "Permit application Ada Example - verified", "purpose": "verify application", "recorded_at": NOW + timedelta(minutes=2), "change_reason": "Identity verified.", "idempotency_key": "record-update-1", }, ) self.session.commit() self.assertEqual(2, revised["revision"]) self.assertEqual("process permit application", revised["purpose"]) token = bind_temporal_data_context( TemporalDataContext( validity_mode="at", valid_at=NOW + timedelta(hours=1), recorded_at=NOW + timedelta(minutes=1, seconds=30), ) ) try: historical = self.records.get_record( self.session, self.principal, record_id="record-1" ) self.assertEqual(1, historical["record"]["revision"]) self.assertEqual( "Permit application Ada Example", historical["record"]["title"] ) finally: reset_temporal_data_context(token) with self.assertRaisesRegex(RecordConflictError, "current revision is 2"): self.records.update_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 1, "title": "Stale title", "purpose": "update record", "recorded_at": NOW + timedelta(minutes=3), "change_reason": "Stale edit.", "idempotency_key": "record-update-stale", }, ) def test_update_replay_returns_the_original_revision(self) -> None: self._create_record() first_payload = { "expected_revision": 1, "title": "First revised title", "purpose": "correct title", "recorded_at": NOW + timedelta(minutes=2), "change_reason": "First correction.", "idempotency_key": "record-update-replay", } first = self.records.update_record( self.session, self.principal, record_id="record-1", payload=first_payload, ) self.records.update_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 2, "title": "Second revised title", "purpose": "correct title again", "recorded_at": NOW + timedelta(minutes=3), "change_reason": "Second correction.", "idempotency_key": "record-update-second", }, ) self.session.commit() replay = self.records.update_record( self.session, self.principal, record_id="record-1", payload=first_payload, ) self.assertEqual(2, first["revision"]) self.assertEqual(2, replay["revision"]) self.assertEqual("First revised title", replay["title"]) def test_file_plan_revisions_reject_indirect_parent_cycles(self) -> None: self.records.write_file_plan_node( self.session, self.principal, payload={ "node_id": "plan-child", "parent_node_id": "plan-permits", "code": "10.20.1", "label": "Child", "recorded_at": NOW + timedelta(minutes=1), "idempotency_key": "catalog-child", }, ) self.records.write_file_plan_node( self.session, self.principal, payload={ "node_id": "plan-grandchild", "parent_node_id": "plan-child", "code": "10.20.1.1", "label": "Grandchild", "recorded_at": NOW + timedelta(minutes=2), "idempotency_key": "catalog-grandchild", }, ) with self.assertRaisesRegex(ValueError, "create a cycle"): self.records.write_file_plan_node( self.session, self.principal, payload={ "node_id": "plan-permits", "parent_node_id": "plan-grandchild", "code": "10.20", "label": "Permits", "recorded_at": NOW + timedelta(minutes=3), "expected_revision": 1, "idempotency_key": "catalog-cycle", }, ) def test_exact_source_filing_is_authorized_idempotent_and_reconstructable( self, ) -> None: self._create_record() request = RecordFilingRequest( tenant_id="tenant-1", record_id="record-1", source=RecordSourceLocator( tenant_id="tenant-1", source_module="files", resource_type="file_version", resource_id="file-1", source_revision="version-7", ), purpose="document decision basis", filing_reason="Submitted evidence belongs to the permit record.", idempotency_key="filing-1", institutional_context={"case_id": "case-1"}, ) first = self.records.file(self.session, self.principal, request=request) self.session.commit() second = self.records.file(self.session, self.principal, request=request) self.assertEqual(first.item_id, second.item_id) self.assertTrue(second.replayed) detail = self.records.get_record( self.session, self.principal, record_id="record-1" ) self.assertEqual("version-7", detail["items"][0]["source"]["source_revision"]) self.assertEqual("a" * 64, detail["items"][0]["content_sha256"]) self.assertEqual( ["record.item_filed", "record.created"], [entry["event_type"] for entry in detail["chronology"]], ) def test_filing_rejects_a_volume_from_another_record(self) -> None: self._create_record() second = self.records.create_record( self.session, self.principal, payload={ "record_id": "record-2", "record_number": "2026/0002", "class_id": "class-permit", "file_plan_node_id": "plan-permits", "title": "Second permit record", "state": "open", "purpose": "process second permit", "recorded_at": NOW + timedelta(minutes=2), "change_reason": "Second application received.", "idempotency_key": "record-create-2", }, ) volume = self.records.create_volume( self.session, self.principal, record_id=str(second["record_id"]), payload={ "volume_id": "volume-2", "label": "Second record volume", "recorded_at": NOW + timedelta(minutes=3), "purpose": "structure record", "idempotency_key": "volume-create-2", }, ) request = RecordFilingRequest( tenant_id="tenant-1", record_id="record-1", volume_id=str(volume["volume_id"]), source=RecordSourceLocator( tenant_id="tenant-1", source_module="files", resource_type="file_version", resource_id="file-1", source_revision="version-7", ), purpose="document decision basis", filing_reason="Attempted cross-record volume filing.", idempotency_key="filing-cross-volume", ) with self.assertRaisesRegex(ValueError, "does not belong to this record"): self.records.file(self.session, self.principal, request=request) def test_tenant_isolation_and_source_discovery(self) -> None: self._create_record() records, total = self.records.list_records(self.session, Principal("tenant-2")) self.assertEqual([], records) self.assertEqual(0, total) self.assertEqual( [ { "id": "files", "source_module": "files", "resource_types": ["file_version"], } ], self.records.source_providers(self.session, self.principal), ) def test_governed_lifecycle_hold_disposition_and_transfer_simulation(self) -> None: fixture = json.loads( ( Path(__file__).parent / "fixtures/service_to_decision_journey.json" ).read_text(encoding="utf-8") ) expected = fixture["expected"] record = self._create_record() self.assertEqual(fixture["record"]["record_id"], record["record_id"]) self.assertEqual(fixture["record"]["record_number"], record["record_number"]) self.assertEqual(fixture["record"]["class_id"], record["class_id"]) closed = self.records.close_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 1, "purpose": "close completed case record", "reason": "The administrative decision is final.", "recorded_at": NOW + timedelta(minutes=2), "idempotency_key": "record-close-1", }, ) self.assertEqual(expected["closed_state"], closed["state"]) self.assertEqual( NOW + timedelta(minutes=2, days=3650), datetime.fromisoformat(str(closed["retention_due_at"])), ) appraised = self.records.appraise_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 2, "outcome": "transfer", "purpose": "appraise completed record", "reason": "Transfer to the institutional archive after review.", "policy_refs": ["records-policy:v1"], "override_retention_not_due": True, "recorded_at": NOW + timedelta(minutes=3), "idempotency_key": "record-appraise-1", }, ) self.assertEqual(expected["appraised_state"], appraised["state"]) hold = self.records.apply_hold( self.session, self.principal, record_id="record-1", payload={ "expected_record_revision": 3, "reason": "Pending judicial review.", "authority": "Court order 2026-17", "purpose": "preserve evidence", "recorded_at": NOW + timedelta(minutes=4), "idempotency_key": "record-hold-1", }, ) with self.assertRaisesRegex(RecordConflictError, "active record hold"): self.records.propose_disposition( self.session, self.principal, record_id="record-1", payload={ "expected_record_revision": 3, "action": "transfer", "reason": "Transfer after retention.", "purpose": "dispose record", "recorded_at": NOW + timedelta(minutes=5), "idempotency_key": "record-disposition-blocked", }, ) self.records.release_hold( self.session, self.principal, record_id="record-1", hold_id=str(hold["hold_id"]), payload={ "expected_hold_revision": 1, "reason": "Judicial review is complete.", "purpose": "resume disposition", "recorded_at": NOW + timedelta(minutes=6), "idempotency_key": "record-hold-release-1", }, ) disposition = self.records.propose_disposition( self.session, self.principal, record_id="record-1", payload={ "disposition_id": "disposition-1", "expected_record_revision": 3, "action": "transfer", "reason": "Transfer after independent review.", "purpose": "dispose record", "policy_refs": ["records-policy:v1"], "recorded_at": NOW + timedelta(minutes=7), "idempotency_key": "record-disposition-1", }, ) self.assertEqual("review_pending", disposition["status"]) self.assertFalse(disposition["consequence_preview"]["external_effect"]) self.records.registry.approvals.approved = True finalized = self.records.finalize_disposition( self.session, self.principal, record_id="record-1", disposition_id="disposition-1", payload={ "expected_disposition_revision": 1, "purpose": "approve disposition", "recorded_at": NOW + timedelta(minutes=8), "idempotency_key": "record-disposition-finalize-1", }, ) self.assertEqual(expected["approved_state"], finalized["record"]["state"]) package = self.records.prepare_transfer_package( self.session, self.principal, record_id="record-1", payload={ "package_id": "package-1", "disposition_id": "disposition-1", "expected_record_revision": 4, "provider_id": "simulation", "profile": "govoplan-simulation-v1", "purpose": "validate archive transfer", "recorded_at": NOW + timedelta(minutes=9), "idempotency_key": "record-package-1", }, ) self.assertEqual("prepared", package["status"]) receipt = self.records.dispatch_transfer_package( self.session, self.principal, record_id="record-1", package_id="package-1", payload={ "expected_package_revision": 1, "purpose": "validate archive transfer", "recorded_at": NOW + timedelta(minutes=10), "idempotency_key": "record-package-dispatch-1", }, ) self.session.commit() self.assertEqual(expected["transfer_status"], receipt["status"]) self.assertTrue(receipt["simulated"]) self.assertEqual( expected["custody_transferred"], receipt["receipt"]["metadata"]["custody_transferred"], ) detail = self.records.get_record( self.session, self.principal, record_id="record-1" ) self.assertEqual("released", detail["holds"][0]["status"]) self.assertEqual( expected["disposition_status"], detail["dispositions"][0]["status"] ) self.assertEqual("simulated_accepted", detail["transfer_packages"][0]["status"]) unknown_package = self.records.prepare_transfer_package( self.session, self.principal, record_id="record-1", payload={ "package_id": "package-unknown", "disposition_id": "disposition-1", "expected_record_revision": 4, "provider_id": "unknown_simulation", "profile": "govoplan-unknown-outcome-v1", "purpose": "exercise unknown transfer recovery", "recorded_at": NOW + timedelta(minutes=11), "idempotency_key": "record-package-unknown", }, ) unknown_receipt = self.records.dispatch_transfer_package( self.session, self.principal, record_id="record-1", package_id="package-unknown", payload={ "expected_package_revision": unknown_package["revision"], "purpose": "exercise unknown transfer recovery", "recorded_at": NOW + timedelta(minutes=12), "idempotency_key": "record-package-unknown-dispatch", }, ) self.assertEqual("outcome_unknown", unknown_receipt["status"]) self.assertFalse(unknown_receipt["receipt"]["retry_safe"]) with self.assertRaisesRegex( ValueError, "unknown outcomes require reconciliation" ): self.records.dispatch_transfer_package( self.session, self.principal, record_id="record-1", package_id="package-unknown", payload={ "expected_package_revision": 2, "purpose": "must not retry unknown transfer", "recorded_at": NOW + timedelta(minutes=13), "idempotency_key": "record-package-unknown-retry", }, ) self.assertEqual( 1, self.records.registry.unknown_archive_provider.dispatch_count ) self.assertEqual( "transfer_pending", self.records.get_record(self.session, self.principal, record_id="record-1")[ "record" ]["state"], ) recovery = self.records.recovery_status( self.session, self.principal, record_id="record-1" ) self.assertTrue(recovery["healthy"]) self.assertTrue(recovery["package_checks"][0]["manifest_verified"]) def test_disposition_waits_for_optional_approvals_and_can_be_withdrawn( self, ) -> None: self.records.registry.approvals = None self._create_record() self.records.close_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 1, "purpose": "close record", "reason": "Work completed.", "recorded_at": NOW + timedelta(minutes=2), "idempotency_key": "close-without-approvals", }, ) self.records.appraise_record( self.session, self.principal, record_id="record-1", payload={ "expected_revision": 2, "outcome": "retain", "purpose": "appraise record", "reason": "Retain permanently.", "override_retention_not_due": True, "recorded_at": NOW + timedelta(minutes=3), "idempotency_key": "appraise-without-approvals", }, ) proposal = self.records.propose_disposition( self.session, self.principal, record_id="record-1", payload={ "expected_record_revision": 3, "action": "retain", "purpose": "dispose record", "reason": "Retain permanently.", "recorded_at": NOW + timedelta(minutes=4), "idempotency_key": "proposal-without-approvals", }, ) self.assertEqual("review_unavailable", proposal["status"]) with self.assertRaisesRegex(ValueError, "not awaiting an available approval"): self.records.finalize_disposition( self.session, self.principal, record_id="record-1", disposition_id=str(proposal["disposition_id"]), payload={ "expected_disposition_revision": 1, "purpose": "finalize disposition", "recorded_at": NOW + timedelta(minutes=5), "idempotency_key": "finalize-without-approvals", }, ) withdrawn = self.records.withdraw_disposition( self.session, self.principal, record_id="record-1", disposition_id=str(proposal["disposition_id"]), payload={ "expected_disposition_revision": 1, "purpose": "correct disposition", "reason": "The appraisal evidence needs correction.", "recorded_at": NOW + timedelta(minutes=6), "idempotency_key": "withdraw-without-approvals", }, ) self.assertEqual("withdrawn", withdrawn["status"]) replacement = self.records.propose_disposition( self.session, self.principal, record_id="record-1", payload={ "expected_record_revision": 3, "action": "retain", "purpose": "dispose record", "reason": "Corrected permanent-retention proposal.", "recorded_at": NOW + timedelta(minutes=7), "idempotency_key": "replacement-without-approvals", }, ) self.assertEqual("review_unavailable", replacement["status"]) if __name__ == "__main__": unittest.main()