90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.files import ManagedArtifactStore, ManagedArtifactWriteRequest
|
|
from govoplan_files.backend.capabilities import FilesArtifactStore
|
|
|
|
|
|
def principal() -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset({"files:file:upload"}),
|
|
),
|
|
account=object(),
|
|
user=SimpleNamespace(id="user-1"),
|
|
)
|
|
|
|
|
|
class _Session:
|
|
def query(self):
|
|
raise AssertionError("not used")
|
|
|
|
def flush(self):
|
|
raise AssertionError("not used")
|
|
|
|
|
|
def stored():
|
|
return SimpleNamespace(
|
|
asset=SimpleNamespace(
|
|
id="file-1",
|
|
display_path="Generated/Templates/result.html",
|
|
owner_type="user",
|
|
),
|
|
version=SimpleNamespace(
|
|
id="version-1",
|
|
filename_at_upload="result.html",
|
|
content_type="text/html",
|
|
size_bytes=6,
|
|
checksum_sha256="0" * 64,
|
|
),
|
|
)
|
|
|
|
|
|
class FilesArtifactStoreTests(unittest.TestCase):
|
|
def test_plain_write_uses_files_authority_and_returns_provider_neutral_ref(self) -> None:
|
|
service = FilesArtifactStore()
|
|
self.assertIsInstance(service, ManagedArtifactStore)
|
|
request = ManagedArtifactWriteRequest(
|
|
filename="result.html",
|
|
payload=b"result",
|
|
content_type="text/html",
|
|
folder="Generated/Templates",
|
|
metadata={"producer_module": "templates"},
|
|
)
|
|
with patch("govoplan_files.backend.capabilities.create_file_asset", return_value=stored()) as create:
|
|
result = service.store_artifact(_Session(), principal(), request=request)
|
|
self.assertEqual("file-1", result.file_asset_id)
|
|
self.assertEqual("version-1", result.file_version_id)
|
|
self.assertEqual("tenant-1", create.call_args.kwargs["tenant_id"])
|
|
self.assertEqual("user-1", create.call_args.kwargs["owner_id"])
|
|
self.assertEqual(b"result", create.call_args.kwargs["data"])
|
|
|
|
def test_idempotent_write_uses_source_provenance_without_payload_metadata(self) -> None:
|
|
request = ManagedArtifactWriteRequest(
|
|
filename="result.html",
|
|
payload=b"secret body",
|
|
content_type="text/html",
|
|
idempotency_key="render-1",
|
|
metadata={"producer_module": "templates", "output_sha256": "a" * 64},
|
|
)
|
|
with patch(
|
|
"govoplan_files.backend.capabilities.sync_file_asset_from_source",
|
|
return_value=(stored(), "unchanged", None),
|
|
) as sync:
|
|
FilesArtifactStore().store_artifact(_Session(), principal(), request=request)
|
|
metadata = sync.call_args.kwargs["metadata"]
|
|
self.assertEqual("render-1", metadata["source_provenance"]["external_id"])
|
|
self.assertNotIn("secret body", str(metadata))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|