feat: expose immutable postbox file references

tests/test_postbox_reference.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from govoplan_core.core.files import PostboxFileReferenceRequest
+from govoplan_files.backend.capabilities import FilesPostboxReferenceService
+from govoplan_files.backend.db.models import FileAsset, FileBlob, FileVersion
+from govoplan_files.backend.storage.common import FileStorageError
+
+
+class _Principal:
+    def __init__(self, *, tenant_id: str = "tenant-1", scopes: set[str] | None = None):
+        self.tenant_id = tenant_id
+        self.user = SimpleNamespace(id="user-1")
+        self._scopes = {"files:file:download"} if scopes is None else scopes
+
+    def has(self, scope: str) -> bool:
+        return scope in self._scopes
+
+
+class _Session:
+    def __init__(self, values: dict[tuple[type[object], str], object]):
+        self.values = values
+
+    def get(self, model: type[object], resource_id: str):
+        return self.values.get((model, resource_id))
+
+
+def _request(reference_type: str = "file_version", reference_id: str = "version-1"):
+    return PostboxFileReferenceRequest(
+        postbox_id="postbox-1",
+        message_id="message-1",
+        reference_type=reference_type,
+        reference_id=reference_id,
+    )
+
+
+def _objects():
+    asset = SimpleNamespace(
+        id="file-1",
+        tenant_id="tenant-1",
+        current_version_id="version-2",
+        deleted_at=None,
+        filename="current.txt",
+        display_path="Evidence/current.txt",
+    )
+    version = SimpleNamespace(
+        id="version-1",
+        tenant_id="tenant-1",
+        file_asset_id="file-1",
+        blob_id="blob-1",
+        filename_at_upload="evidence.txt",
+        content_type="text/plain",
+        size_bytes=8,
+        checksum_sha256="a" * 64,
+    )
+    blob = SimpleNamespace(
+        id="blob-1",
+        tenant_id="tenant-1",
+        content_type="text/plain",
+    )
+    return asset, version, blob
+
+
+class FilesPostboxReferenceTests(unittest.TestCase):
+    def test_resolves_the_referenced_version_instead_of_the_current_version(self) -> None:
+        asset, version, blob = _objects()
+        session = _Session(
+            {
+                (FileAsset, asset.id): asset,
+                (FileVersion, version.id): version,
+                (FileBlob, blob.id): blob,
+            }
+        )
+
+        with patch(
+            "govoplan_files.backend.capabilities.get_asset_for_user",
+            return_value=asset,
+        ):
+            result = FilesPostboxReferenceService().resolve_postbox_references(
+                session,
+                _Principal(),
+                tenant_id="tenant-1",
+                requests=(_request(),),
+            )[0]
+
+        self.assertTrue(result.available)
+        self.assertEqual("version-1", result.file_version_id)
+        self.assertEqual(
+            "/api/v1/files/file-1/versions/version-1/download",
+            result.download_path,
+        )
+        self.assertTrue(result.provenance["exact_version"])
+
+    def test_fails_closed_for_tenant_permission_and_file_access_mismatches(self) -> None:
+        service = FilesPostboxReferenceService()
+        empty = _Session({})
+
+        tenant_result = service.resolve_postbox_references(
+            empty,
+            _Principal(tenant_id="tenant-2"),
+            tenant_id="tenant-1",
+            requests=(_request(),),
+        )[0]
+        permission_result = service.resolve_postbox_references(
+            empty,
+            _Principal(scopes=set()),
+            tenant_id="tenant-1",
+            requests=(_request(),),
+        )[0]
+
+        asset, version, blob = _objects()
+        populated = _Session(
+            {
+                (FileAsset, asset.id): asset,
+                (FileVersion, version.id): version,
+                (FileBlob, blob.id): blob,
+            }
+        )
+        with patch(
+            "govoplan_files.backend.capabilities.get_asset_for_user",
+            side_effect=FileStorageError("denied"),
+        ):
+            access_result = service.resolve_postbox_references(
+                populated,
+                _Principal(),
+                tenant_id="tenant-1",
+                requests=(_request(),),
+            )[0]
+
+        self.assertEqual("tenant_mismatch", tenant_result.reason_code)
+        self.assertEqual("download_permission_missing", permission_result.reason_code)
+        self.assertEqual("file_access_denied", access_result.reason_code)
+        self.assertFalse(access_result.available)
+
+    def test_asset_reference_does_not_drift_to_a_later_current_version(self) -> None:
+        asset, version, blob = _objects()
+        session = _Session(
+            {
+                (FileAsset, asset.id): asset,
+                (FileVersion, version.id): version,
+                (FileBlob, blob.id): blob,
+            }
+        )
+
+        result = FilesPostboxReferenceService().resolve_postbox_references(
+            session,
+            _Principal(),
+            tenant_id="tenant-1",
+            requests=(_request("file", asset.id),),
+        )[0]
+
+        self.assertFalse(result.available)
+        self.assertEqual("exact_version_required", result.reason_code)
+
+
+if __name__ == "__main__":
+    unittest.main()