from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime, timedelta import hashlib import json from pathlib import Path import sqlite3 import tempfile 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 ReferenceJourneySourceProvider: def __init__(self, provider_id: str, resource_types: tuple[str, ...]) -> None: self.provider_id = provider_id self._resource_types = resource_types def resource_types(self): return self._resource_types def resolve(self, session, principal, *, locator, purpose): del session, purpose if principal.tenant_id != locator.tenant_id: raise ValueError("Source access denied.") if locator.source_module != self.provider_id: raise ValueError("The source module does not match this provider.") if locator.resource_type not in self._resource_types: raise ValueError("The source type is not supported by this provider.") metadata = dict(locator.metadata) reference = dict(metadata.get("reference") or {}) digest_input = ":".join( ( locator.source_module, locator.resource_type, locator.resource_id, locator.source_revision, ) ).encode("utf-8") return RecordSourceReference( locator=locator, label=str(reference.get("label") or locator.resource_id), authority_mode=str( reference.get("authority_mode") or "linked_reference" ), content_sha256=hashlib.sha256(digest_input).hexdigest(), content_type=str(reference.get("content_type") or "application/json"), size_bytes=int(reference.get("size_bytes") or 0), valid_from=NOW, recorded_at=NOW, launch_url=str(reference.get("launch_url") or "") or None, metadata=metadata, ) 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 ReferenceJourneyRegistry(Registry): def __init__(self) -> None: super().__init__() self.source_providers_by_module = { "files": ReferenceJourneySourceProvider("files", ("file_version",)), "forms_runtime": ReferenceJourneySourceProvider( "forms_runtime", ("form_submission_revision",) ), "cases": ReferenceJourneySourceProvider("cases", ("case_revision",)), "decisions": ReferenceJourneySourceProvider( "decisions", ("decision_revision",) ), } def capability_names(self): return ( *(f"records.source.{name}" for name in self.source_providers_by_module), "records.archive.simulation", "records.archive.unknown_simulation", ) def tenant_capability(self, name, session, *, tenant_id): if name.startswith("records.source.") and tenant_id == "tenant-1": return self.source_providers_by_module.get(name.removeprefix("records.source.")) return super().tenant_capability(name, session, tenant_id=tenant_id) 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", "forms_runtime:form_submission_revision", "cases:case_revision", "decisions:decision_revision", ], "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_assisted_service_reference_journey_survives_backup_and_restore( self, ) -> None: fixture = json.loads( ( Path(__file__).parent / "fixtures/service_to_decision_journey.json" ).read_text(encoding="utf-8") ) expected = fixture["expected"] context = fixture["institutional_context"] self.records = SqlRecordRegistry(ReferenceJourneyRegistry()) def create_journey_record( record_data: dict[str, object], record_context: dict[str, object], *, idempotency_suffix: str, channel: str, ) -> dict[str, object]: return self.records.create_record( self.session, self.principal, payload={ **record_data, "file_plan_node_id": "plan-permits", "description": fixture["description"], "state": "open", "source_authority_mode": "native_authoritative", "access_mode": "tenant", "purpose": record_context["purpose"], "responsible_unit_id": "unit-mobility-services", "responsible_function_id": record_context[ "responsible_function_id" ], "institutional_context": {**record_context, "channel": channel}, "recorded_at": NOW + timedelta(minutes=1), "valid_from": NOW, "change_reason": f"The {channel} application was accepted for processing.", "idempotency_key": f"reference-record-create-{idempotency_suffix}", }, ) def filing_request( record_id: str, source: dict[str, object], record_context: dict[str, object], *, idempotency_suffix: str = "", ) -> RecordFilingRequest: metadata = dict(source["metadata"]) return RecordFilingRequest( tenant_id="tenant-1", record_id=record_id, source=RecordSourceLocator( tenant_id="tenant-1", source_module=str(source["source_module"]), resource_type=str(source["resource_type"]), resource_id=str(source["resource_id"]), source_revision=str(source["source_revision"]), metadata={ "reference": { "label": source["label"], "authority_mode": source["authority_mode"], "content_type": source["content_type"], "size_bytes": source["size_bytes"], "launch_url": source["launch_url"], }, **metadata, }, ), purpose=str(source["purpose"]), filing_reason=str(source["filing_reason"]), relationship=str(source["relationship"]), institutional_context={ **record_context, "evidence_role": metadata["evidence_role"], }, metadata={"fixture_source_id": source["id"]}, idempotency_key=( f"reference-file-{source['id']}{idempotency_suffix}" ), ) record = create_journey_record( fixture["record"], context, idempotency_suffix="assisted", channel="assisted" ) self.assertEqual(fixture["record"]["title"], record["title"]) filed_results = [] for source in fixture["sources"]: filed_results.append( self.records.file( self.session, self.principal, request=filing_request(str(record["record_id"]), source, context), ) ) replay = self.records.file( self.session, self.principal, request=filing_request( str(record["record_id"]), fixture["sources"][0], context ), ) self.assertTrue(replay.replayed) self.assertEqual(filed_results[0].item_id, replay.item_id) detail = self.records.get_record( self.session, self.principal, record_id=str(record["record_id"]) ) self.assertEqual(expected["filed_item_count"], len(detail["items"])) self.assertEqual( [source["source_revision"] for source in fixture["sources"]], [item["source"]["source_revision"] for item in detail["items"]], ) self.assertEqual( [source["relationship"] for source in fixture["sources"]], [item["relationship"] for item in detail["items"]], ) self.assertEqual( { "application", "application_attachment", "case_context", "formal_decision", "delivery_receipt", "correction", }, {item["source_metadata"]["evidence_role"] for item in detail["items"]}, ) digital_fixture = fixture["digital_equivalent"] digital_context = { **context, **digital_fixture["institutional_context_overrides"], } digital_record = create_journey_record( digital_fixture["record"], digital_context, idempotency_suffix="digital", channel="digital", ) digital_sources = [digital_fixture["intake_source"]] for source in fixture["sources"][1:]: digital_sources.append( { **source, "id": f"digital-{source['id']}", "resource_id": f"{source['resource_id']}-digital", "label": f"Digital equivalent ยท {source['label']}", } ) for source in digital_sources: self.records.file( self.session, self.principal, request=filing_request( str(digital_record["record_id"]), source, digital_context, idempotency_suffix="-digital", ), ) digital_detail = self.records.get_record( self.session, self.principal, record_id=str(digital_record["record_id"]), ) def equivalence_fields(item): return ( item["source"]["source_module"], item["source"]["resource_type"], item["relationship"], item["source_metadata"]["evidence_role"], ) self.assertEqual( [equivalence_fields(item) for item in detail["items"]], [equivalence_fields(item) for item in digital_detail["items"]], ) all_records, total_records = self.records.list_records( self.session, self.principal ) self.assertEqual(expected["equivalent_record_count"], total_records) self.assertEqual( {fixture["record"]["record_id"], digital_fixture["record"]["record_id"]}, {item["record_id"] for item in all_records}, ) lifecycle_start = datetime.now(UTC).replace(microsecond=0) + timedelta( minutes=10 ) closed = self.records.close_record( self.session, self.principal, record_id=str(record["record_id"]), payload={ "expected_revision": 1, "purpose": "close completed resident parking permit record", "reason": "Decision, delivery, and correction evidence are complete.", "recorded_at": lifecycle_start, "idempotency_key": "reference-close", }, ) self.assertEqual(expected["closed_state"], closed["state"]) appraised = self.records.appraise_record( self.session, self.principal, record_id=str(record["record_id"]), payload={ "expected_revision": 2, "outcome": "transfer", "purpose": "appraise completed resident parking permit record", "reason": "Offer the complete record after governed review.", "policy_refs": [context["retention_policy_ref"]], "override_retention_not_due": True, "recorded_at": lifecycle_start + timedelta(minutes=1), "idempotency_key": "reference-appraise", }, ) self.assertEqual(expected["appraised_state"], appraised["state"]) hold = self.records.apply_hold( self.session, self.principal, record_id=str(record["record_id"]), payload={ "expected_record_revision": 3, "reason": "Preserve the record while a correction is reviewed.", "authority": "Mobility authority review 2026-17", "purpose": "preserve correction evidence", "policy_refs": [context["retention_policy_ref"]], "institutional_context": context, "recorded_at": lifecycle_start + timedelta(minutes=2), "idempotency_key": "reference-hold", }, ) with self.assertRaisesRegex(RecordConflictError, "active record hold"): self.records.propose_disposition( self.session, self.principal, record_id=str(record["record_id"]), payload={ "expected_record_revision": 3, "action": "transfer", "reason": "Attempt transfer while evidence is held.", "purpose": "prove hold enforcement", "recorded_at": lifecycle_start + timedelta(minutes=3), "idempotency_key": "reference-disposition-blocked", }, ) self.records.release_hold( self.session, self.principal, record_id=str(record["record_id"]), hold_id=str(hold["hold_id"]), payload={ "expected_hold_revision": 1, "reason": "Correction review completed with prior evidence preserved.", "purpose": "resume governed disposition", "recorded_at": lifecycle_start + timedelta(minutes=4), "idempotency_key": "reference-hold-release", }, ) disposition = self.records.propose_disposition( self.session, self.principal, record_id=str(record["record_id"]), payload={ "disposition_id": "reference-disposition", "expected_record_revision": 3, "action": "transfer", "reason": "The complete record is ready for an independently reviewed offer.", "purpose": "prepare governed archive offer", "policy_refs": [context["retention_policy_ref"]], "institutional_context": context, "recorded_at": lifecycle_start + timedelta(minutes=5), "idempotency_key": "reference-disposition", }, ) self.records.registry.approvals.approved = True finalized = self.records.finalize_disposition( self.session, self.principal, record_id=str(record["record_id"]), disposition_id=str(disposition["disposition_id"]), payload={ "expected_disposition_revision": 1, "purpose": "record independent disposition approval", "recorded_at": lifecycle_start + timedelta(minutes=6), "idempotency_key": "reference-disposition-finalize", }, ) self.assertEqual(expected["approved_state"], finalized["record"]["state"]) package = self.records.prepare_transfer_package( self.session, self.principal, record_id=str(record["record_id"]), payload={ "package_id": "reference-transfer-package", "disposition_id": str(disposition["disposition_id"]), "expected_record_revision": 4, "provider_id": "simulation", "profile": "govoplan-simulation-v1", "purpose": "validate archive transfer boundary", "recorded_at": lifecycle_start + timedelta(minutes=7), "idempotency_key": "reference-transfer-prepare", }, ) receipt = self.records.dispatch_transfer_package( self.session, self.principal, record_id=str(record["record_id"]), package_id=str(package["package_id"]), payload={ "expected_package_revision": 1, "purpose": "validate archive receipt handling", "recorded_at": lifecycle_start + timedelta(minutes=8), "idempotency_key": "reference-transfer-dispatch", }, ) self.assertEqual(expected["transfer_status"], receipt["status"]) self.assertEqual( expected["custody_transferred"], receipt["receipt"]["metadata"]["custody_transferred"], ) self.session.commit() with tempfile.TemporaryDirectory(prefix="govoplan-records-restore-") as temp_dir: backup_path = Path(temp_dir) / "restored-records.sqlite3" source_connection = self.engine.raw_connection() try: with sqlite3.connect(backup_path) as backup_connection: source_connection.driver_connection.backup(backup_connection) finally: source_connection.close() restored_engine = create_engine(f"sqlite+pysqlite:///{backup_path}") restored_session = Session(restored_engine) try: restored_records = SqlRecordRegistry(ReferenceJourneyRegistry()) matches, total = restored_records.list_records( restored_session, self.principal, query=expected["restored_search_query"], ) self.assertEqual(1, total) self.assertEqual(record["record_id"], matches[0]["record_id"]) restored = restored_records.get_record( restored_session, self.principal, record_id=str(record["record_id"]), ) self.assertEqual(expected["filed_item_count"], len(restored["items"])) self.assertEqual( {**context, "channel": "assisted"}, restored["record"]["institutional_context"], ) self.assertEqual("released", restored["holds"][0]["status"]) self.assertEqual( expected["disposition_status"], restored["dispositions"][0]["status"], ) self.assertEqual( [context["retention_policy_ref"]], restored["dispositions"][0]["policy_refs"], ) self.assertEqual( expected["transfer_status"], restored["transfer_packages"][0]["status"], ) self.assertTrue(restored["transfer_packages"][0]["simulated"]) self.assertEqual( expected["filed_item_count"], sum( entry["event_type"] == "record.item_filed" for entry in restored["chronology"] ), ) recovery = restored_records.recovery_status( restored_session, self.principal, record_id=str(record["record_id"]), ) self.assertTrue(recovery["healthy"]) self.assertEqual( expected["filed_item_count"], len(recovery["source_checks"]) ) self.assertTrue( all( check["status"] == "verified" for check in recovery["source_checks"] ) ) self.assertTrue( all( check["manifest_verified"] for check in recovery["package_checks"] ) ) token = bind_temporal_data_context( TemporalDataContext( validity_mode="at", valid_at=lifecycle_start - timedelta(minutes=1), recorded_at=lifecycle_start - timedelta(minutes=1), ) ) try: historical = restored_records.get_record( restored_session, self.principal, record_id=str(record["record_id"]), ) self.assertEqual("open", historical["record"]["state"]) self.assertEqual( expected["filed_item_count"], len(historical["items"]) ) self.assertEqual([], historical["transfer_packages"]) finally: reset_temporal_data_context(token) finally: restored_session.close() restored_engine.dispose() 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()