161 lines
5.1 KiB
Python
161 lines
5.1 KiB
Python
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()
|