feat(cases): expose exact revisions for records filing

2026-08-06 01:43:32 +02:00
parent cfe2389e5a
commit 2f030e56a9
4 changed files with 284 additions and 0 deletions
@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.records import RecordContractError, RecordSourceLocator
+from govoplan_cases.backend.db.models import CaseIdentity, CaseRecordRevision
+from govoplan_cases.backend.record_source import CasesRecordSource
+
+
+NOW = datetime(2026, 1, 6, 9, 0, tzinfo=UTC)
+
+
+@dataclass
+class Principal:
+    tenant_id: str = "tenant-1"
+
+    def has(self, scope: str) -> bool:
+        return scope == "cases:case:read"
+
+
+class CasesRecordSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        CaseIdentity.__table__.create(self.engine)
+        CaseRecordRevision.__table__.create(self.engine)
+        self.session = Session(self.engine)
+        identity = CaseIdentity(
+            id="identity-1",
+            tenant_id="tenant-1",
+            case_id="case-1",
+            case_number="2026/C-1",
+            created_by="account-1",
+        )
+        revision = CaseRecordRevision(
+            id="revision-1",
+            tenant_id="tenant-1",
+            case_id="case-1",
+            identity_id="identity-1",
+            revision=3,
+            case_type_key="permit",
+            status_key="review",
+            title="Permit application",
+            access_mode="restricted",
+            search_text="permit application",
+            opened_at=NOW,
+            recorded_at=NOW,
+            snapshot={
+                "case_id": "case-1",
+                "revision": 3,
+                "title": "Permit application",
+            },
+        )
+        self.session.add_all((identity, revision))
+        self.session.flush()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def locator(self) -> RecordSourceLocator:
+        return RecordSourceLocator(
+            tenant_id="tenant-1",
+            source_module="cases",
+            resource_type="case_revision",
+            resource_id="case-1",
+            source_revision="3",
+        )
+
+    def test_resolves_exact_currently_authorized_case_revision(self) -> None:
+        with patch(
+            "govoplan_cases.backend.record_source.can_access_case", return_value=True
+        ):
+            result = CasesRecordSource().resolve(
+                self.session,
+                Principal(),
+                locator=self.locator(),
+                purpose="preserve decision basis",
+            )
+
+        self.assertEqual("2026/C-1 - Permit application", result.label)
+        self.assertEqual(64, len(result.content_sha256 or ""))
+        self.assertEqual("review", result.metadata["status_key"])
+
+    def test_object_access_is_rechecked_and_fails_closed(self) -> None:
+        with (
+            patch(
+                "govoplan_cases.backend.record_source.can_access_case",
+                return_value=False,
+            ),
+            self.assertRaisesRegex(RecordContractError, "cannot read"),
+        ):
+            CasesRecordSource().resolve(
+                self.session,
+                Principal(),
+                locator=self.locator(),
+                purpose="preserve decision basis",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()