feat(files): expose exact versions for records filing

tests/test_manifest_documentation.py

@@ -14,6 +14,7 @@ STATIC_TOPIC_IDS = {
     "files.reference.shared-storage-profile",
     "files.reference.generated-artifact-store",
     "files.reference.snapshot-provenance-and-capabilities",
+    "files.records.exact-version-source",
     "files.assurance.process-and-release-readiness",
 }
 RUNTIME_TOPIC_IDS = {

tests/test_record_source.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from types import SimpleNamespace
+import unittest
+from unittest.mock import MagicMock, patch
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.records import RecordContractError, RecordSourceLocator
+from govoplan_files.backend.db.models import FileAsset, FileBlob, FileVersion
+from govoplan_files.backend.record_source import FilesRecordSource
+
+
+@dataclass
+class Principal:
+    tenant_id: str = "tenant-1"
+    membership_id: str = "user-1"
+    user: object = field(default_factory=lambda: SimpleNamespace(id="user-1"))
+
+    def has(self, scope: str) -> bool:
+        return scope in {"files:file:read"}
+
+
+class FilesRecordSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.session = MagicMock(spec=Session)
+        self.asset = FileAsset(
+            id="asset-1",
+            tenant_id="tenant-1",
+            owner_type="user",
+            owner_user_id="user-1",
+            current_version_id="version-1",
+            display_path="Evidence/Decision.pdf",
+            filename="Decision.pdf",
+        )
+        self.blob = FileBlob(
+            id="blob-1",
+            tenant_id="tenant-1",
+            storage_backend="local",
+            storage_key="tenant-1/blob-1",
+            checksum_sha256="a" * 64,
+            size_bytes=1024,
+            integrity_status="verified",
+        )
+        self.version = FileVersion(
+            id="version-1",
+            tenant_id="tenant-1",
+            file_asset_id="asset-1",
+            blob_id="blob-1",
+            version_number=1,
+            filename_at_upload="Decision.pdf",
+            display_path_at_upload="Evidence/Decision.pdf",
+            content_type="application/pdf",
+            size_bytes=1024,
+            checksum_sha256="a" * 64,
+        )
+        self.version.created_at = datetime.now(UTC)
+        version_query = self.session.query.return_value.filter.return_value
+        version_query.filter.return_value.one_or_none.return_value = self.version
+        self.session.get.return_value = self.blob
+
+    def locator(self) -> RecordSourceLocator:
+        return RecordSourceLocator(
+            tenant_id="tenant-1",
+            source_module="files",
+            resource_type="file_version",
+            resource_id="asset-1",
+            source_revision="version-1",
+        )
+
+    def test_resolves_exact_authorized_version_and_digest(self) -> None:
+        with patch(
+            "govoplan_files.backend.record_source.get_asset_for_user",
+            return_value=self.asset,
+        ):
+            result = FilesRecordSource().resolve(
+                self.session,
+                Principal(),
+                locator=self.locator(),
+                purpose="document decision basis",
+            )
+
+        self.assertEqual("version-1", result.locator.source_revision)
+        self.assertEqual("a" * 64, result.content_sha256)
+        self.assertEqual("verified", result.metadata["integrity_status"])
+
+    def test_quarantined_version_fails_closed(self) -> None:
+        self.blob.quarantined_at = datetime.now(UTC)
+        self.session.flush()
+        with (
+            patch(
+                "govoplan_files.backend.record_source.get_asset_for_user",
+                return_value=self.asset,
+            ),
+            self.assertRaisesRegex(RecordContractError, "integrity gate"),
+        ):
+            FilesRecordSource().resolve(
+                self.session,
+                Principal(),
+                locator=self.locator(),
+                purpose="document decision basis",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()