165 lines
5.4 KiB
Python
165 lines
5.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.db.models import Account, Group, User
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_files.backend.db.models import (
|
|
FileAsset,
|
|
FileBlob,
|
|
FileFolder,
|
|
FileFormEvidenceGrant,
|
|
FileShare,
|
|
FileVersion,
|
|
)
|
|
from govoplan_files.backend.storage.common import FileStorageError, utcnow
|
|
from govoplan_files.backend.storage.lifecycle import (
|
|
preview_asset_purge,
|
|
restore_asset,
|
|
set_asset_lifecycle,
|
|
)
|
|
|
|
|
|
class FileLifecycleTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(
|
|
bind=self.engine,
|
|
tables=[
|
|
Account.__table__,
|
|
User.__table__,
|
|
Group.__table__,
|
|
ChangeSequenceEntry.__table__,
|
|
FileBlob.__table__,
|
|
FileAsset.__table__,
|
|
FileVersion.__table__,
|
|
FileFolder.__table__,
|
|
FileFormEvidenceGrant.__table__,
|
|
FileShare.__table__,
|
|
],
|
|
)
|
|
self.session = sessionmaker(bind=self.engine, future=True)()
|
|
self.blob = FileBlob(
|
|
id="blob-1",
|
|
tenant_id="tenant-1",
|
|
storage_backend="local",
|
|
storage_key="tenants/tenant-1/files/blob-1",
|
|
checksum_sha256="a" * 64,
|
|
size_bytes=1,
|
|
ref_count=1,
|
|
)
|
|
self.asset = FileAsset(
|
|
id="file-1",
|
|
tenant_id="tenant-1",
|
|
owner_type="user",
|
|
owner_user_id="user-1",
|
|
current_version_id="version-1",
|
|
display_path="records/report.txt",
|
|
filename="report.txt",
|
|
deleted_at=utcnow(),
|
|
)
|
|
self.version = FileVersion(
|
|
id="version-1",
|
|
tenant_id="tenant-1",
|
|
file_asset_id="file-1",
|
|
blob_id="blob-1",
|
|
version_number=1,
|
|
filename_at_upload="report.txt",
|
|
display_path_at_upload="records/report.txt",
|
|
size_bytes=1,
|
|
checksum_sha256="a" * 64,
|
|
)
|
|
self.session.add_all([self.blob, self.asset, self.version])
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_preview_hash_changes_with_retention_and_blocks_active_policy(self) -> None:
|
|
first = preview_asset_purge(
|
|
self.session, tenant_id="tenant-1", file_ids=[self.asset.id]
|
|
)
|
|
self.assertTrue(first.eligible)
|
|
|
|
set_asset_lifecycle(
|
|
self.session,
|
|
self.asset,
|
|
retained_until=utcnow() + timedelta(days=7),
|
|
legal_hold=True,
|
|
reason="Active proceeding",
|
|
expected_revision=1,
|
|
)
|
|
self.session.commit()
|
|
second = preview_asset_purge(
|
|
self.session, tenant_id="tenant-1", file_ids=[self.asset.id]
|
|
)
|
|
|
|
self.assertFalse(second.eligible)
|
|
self.assertNotEqual(first.preview_sha256, second.preview_sha256)
|
|
self.assertEqual(
|
|
{"legal_hold", "retention_active"}, set(second.items[0].blockers)
|
|
)
|
|
|
|
def test_restore_preserves_version_and_provenance_and_rejects_collision(self) -> None:
|
|
self.asset.metadata_ = {"source": {"provider": "s3", "revision": "v1"}}
|
|
self.session.commit()
|
|
|
|
self.assertTrue(restore_asset(self.session, self.asset))
|
|
self.assertEqual("version-1", self.asset.current_version_id)
|
|
self.assertEqual("v1", self.asset.metadata_["source"]["revision"])
|
|
self.session.commit()
|
|
|
|
self.asset.deleted_at = utcnow()
|
|
collision = FileAsset(
|
|
id="file-2",
|
|
tenant_id="tenant-1",
|
|
owner_type="user",
|
|
owner_user_id="user-1",
|
|
display_path=self.asset.display_path,
|
|
filename=self.asset.filename,
|
|
)
|
|
self.session.add(collision)
|
|
self.session.commit()
|
|
with self.assertRaisesRegex(FileStorageError, "path is already in use"):
|
|
restore_asset(self.session, self.asset)
|
|
|
|
def test_active_share_blocks_purge_until_revoked(self) -> None:
|
|
share = FileShare(
|
|
id="share-1",
|
|
tenant_id="tenant-1",
|
|
file_asset_id=self.asset.id,
|
|
target_type="tenant",
|
|
target_id="tenant-1",
|
|
permission="read",
|
|
)
|
|
self.session.add(share)
|
|
self.session.commit()
|
|
|
|
blocked = preview_asset_purge(
|
|
self.session, tenant_id="tenant-1", file_ids=[self.asset.id]
|
|
)
|
|
self.assertIn("active_share", blocked.items[0].blockers)
|
|
|
|
share.revoked_at = utcnow()
|
|
self.session.commit()
|
|
allowed = preview_asset_purge(
|
|
self.session, tenant_id="tenant-1", file_ids=[self.asset.id]
|
|
)
|
|
self.assertTrue(allowed.eligible)
|
|
|
|
def test_missing_campaign_table_is_an_optional_module_safe_path(self) -> None:
|
|
preview = preview_asset_purge(
|
|
self.session, tenant_id="tenant-1", file_ids=[self.asset.id]
|
|
)
|
|
self.assertNotIn("campaign_evidence", preview.items[0].blockers)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|