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.records import ( RecordFilingRequest, RecordSourceLocator, RecordSourceReference, ) from govoplan_core.core.temporal import ( TemporalDataContext, bind_temporal_data_context, reset_temporal_data_context, ) from govoplan_records.backend.db.models import ( RecordChronologyEntry, RecordClassRevision, RecordFilePlanRevision, RecordIdentity, RecordItem, RecordRevision, RecordVolumeRevision, ) 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 Registry: def __init__(self) -> None: self.provider = SourceProvider() def capability_names(self): return ("records.source.files",) def tenant_capability(self, name, session, *, tenant_id): del session return ( self.provider if name == "records.source.files" and tenant_id == "tenant-1" else None ) 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__, ): 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), ) if __name__ == "__main__": unittest.main()