109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
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()
|