558 lines
22 KiB
Python
558 lines
22 KiB
Python
# #region Test.FileIO.Utils [C:3] [TYPE Module] [SEMANTICS test,fileio,utils,filename,sanitize,crc32,sha256,manifest,integrity]
|
|
# @BRIEF Tests for core/utils/fileio.py — pure utility functions: sanitize_filename, get_filename_from_headers, calculate_crc32, calculate_sha256, compute_archive_content_hash, write_backup_manifest, read_backup_manifest, verify_backup_integrity, create_temp_file, remove_empty_directories.
|
|
# @RELATION BINDS_TO -> [FileIO]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import os
|
|
import tempfile
|
|
import zipfile
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
class TestSanitizeFilename:
|
|
"""sanitize_filename — removes invalid filename characters."""
|
|
|
|
def test_already_clean(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert sanitize_filename("dashboard_export.zip") == "dashboard_export.zip"
|
|
|
|
def test_replaces_backslash(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert "\\" not in sanitize_filename("a\\b")
|
|
assert "_" in sanitize_filename("a\\b")
|
|
|
|
def test_replaces_colon(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert ":" not in sanitize_filename("a:b")
|
|
assert "_" in sanitize_filename("a:b")
|
|
|
|
def test_replaces_angle_brackets(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert "<" not in sanitize_filename("a<b")
|
|
assert ">" not in sanitize_filename("a>b")
|
|
|
|
def test_replaces_question_mark(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert "?" not in sanitize_filename("a?b")
|
|
assert "_" in sanitize_filename("a?b")
|
|
|
|
def test_replaces_asterisk(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert "*" not in sanitize_filename("a*b")
|
|
assert "_" in sanitize_filename("a*b")
|
|
|
|
def test_replaces_pipe(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert "|" not in sanitize_filename("a|b")
|
|
assert "_" in sanitize_filename("a|b")
|
|
|
|
def test_replaces_quote(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert '"' not in sanitize_filename('a"b')
|
|
assert "_" in sanitize_filename('a"b')
|
|
|
|
def test_strips_whitespace(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
result = sanitize_filename(" file.txt ")
|
|
assert result == "file.txt"
|
|
assert result == result.strip()
|
|
|
|
def test_empty_string(self):
|
|
from src.core.utils.fileio import sanitize_filename
|
|
assert sanitize_filename("") == ""
|
|
|
|
|
|
class TestGetFilenameFromHeaders:
|
|
"""get_filename_from_headers — extracts filename from Content-Disposition."""
|
|
|
|
def test_standard_header(self):
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
headers = {"Content-Disposition": 'attachment; filename="report.pdf"'}
|
|
assert get_filename_from_headers(headers) == "report.pdf"
|
|
|
|
def test_header_without_quotes(self):
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
headers = {"Content-Disposition": "attachment; filename=report.pdf"}
|
|
assert get_filename_from_headers(headers) == "report.pdf"
|
|
|
|
def test_missing_header(self):
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
assert get_filename_from_headers({}) is None
|
|
|
|
def test_empty_header(self):
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
headers = {"Content-Disposition": ""}
|
|
assert get_filename_from_headers(headers) is None
|
|
|
|
def test_no_filename_in_header(self):
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
headers = {"Content-Disposition": "inline"}
|
|
assert get_filename_from_headers(headers) is None
|
|
|
|
def test_filename_with_spaces(self):
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
headers = {"Content-Disposition": 'attachment; filename="my report.csv"'}
|
|
assert get_filename_from_headers(headers) == "my report.csv"
|
|
|
|
def test_utf8_filename_with_regular_format(self):
|
|
"""The regex only matches filename=\"...\", not filename*=UTF-8''..."""
|
|
from src.core.utils.fileio import get_filename_from_headers
|
|
headers = {"Content-Disposition": 'attachment; filename="report.pdf"; filename*=UTF-8\'\'%D0%BE%D1%82%D1%87%D0%B5%D1%82.pdf'}
|
|
result = get_filename_from_headers(headers)
|
|
assert result == "report.pdf"
|
|
|
|
|
|
class TestCalculateCrc32:
|
|
"""calculate_crc32 — computes CRC32 hash of a file."""
|
|
|
|
def test_known_content(self):
|
|
import zlib
|
|
from src.core.utils.fileio import calculate_crc32
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
f.write(b"hello world")
|
|
f.flush()
|
|
fname = f.name
|
|
try:
|
|
crc = calculate_crc32(fname)
|
|
assert isinstance(crc, str)
|
|
assert len(crc) == 8 # 32-bit hex, 8 chars
|
|
expected = format(zlib.crc32(b"hello world") & 0xFFFFFFFF, "08x")
|
|
assert crc == expected
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_empty_file(self):
|
|
from src.core.utils.fileio import calculate_crc32
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
crc = calculate_crc32(fname)
|
|
assert isinstance(crc, str)
|
|
assert len(crc) == 8
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
|
|
class TestCalculateSha256:
|
|
"""calculate_sha256 — computes SHA256 hex digest of a file."""
|
|
|
|
def test_known_content(self):
|
|
import hashlib
|
|
from src.core.utils.fileio import calculate_sha256
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
f.write(b"hello world")
|
|
f.flush()
|
|
fname = f.name
|
|
try:
|
|
sha = calculate_sha256(Path(fname))
|
|
assert isinstance(sha, str)
|
|
assert len(sha) == 64 # SHA256 hex is 64 chars
|
|
expected = hashlib.sha256(b"hello world").hexdigest()
|
|
assert sha == expected
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_empty_file(self):
|
|
from src.core.utils.fileio import calculate_sha256
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
sha = calculate_sha256(Path(fname))
|
|
assert isinstance(sha, str)
|
|
assert len(sha) == 64
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_deterministic(self):
|
|
from src.core.utils.fileio import calculate_sha256
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
f.write(b"same content")
|
|
f.flush()
|
|
fname_a = f.name
|
|
try:
|
|
with tempfile.NamedTemporaryFile(delete=False) as f2:
|
|
f2.write(b"same content")
|
|
f2.flush()
|
|
fname_b = f2.name
|
|
try:
|
|
sha_a = calculate_sha256(Path(fname_a))
|
|
sha_b = calculate_sha256(Path(fname_b))
|
|
assert sha_a == sha_b
|
|
finally:
|
|
os.unlink(fname_b)
|
|
finally:
|
|
os.unlink(fname_a)
|
|
|
|
|
|
class TestComputeArchiveContentHash:
|
|
"""compute_archive_content_hash — deterministic hash of ZIP member manifest."""
|
|
|
|
def test_empty_zip(self):
|
|
from src.core.utils.fileio import compute_archive_content_hash
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w"):
|
|
pass
|
|
h = compute_archive_content_hash(Path(fname))
|
|
assert isinstance(h, str)
|
|
assert len(h) == 64
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_with_members(self):
|
|
from src.core.utils.fileio import compute_archive_content_hash
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("file1.txt", "hello")
|
|
zf.writestr("file2.txt", "world")
|
|
h = compute_archive_content_hash(Path(fname))
|
|
assert isinstance(h, str)
|
|
assert len(h) == 64
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_deterministic(self):
|
|
from src.core.utils.fileio import compute_archive_content_hash
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname_a = f.name
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname_b = f.name
|
|
try:
|
|
for fname in (fname_a, fname_b):
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("a.txt", "data")
|
|
zf.writestr("b.txt", "more data")
|
|
assert compute_archive_content_hash(Path(fname_a)) == compute_archive_content_hash(Path(fname_b))
|
|
finally:
|
|
os.unlink(fname_a)
|
|
os.unlink(fname_b)
|
|
|
|
def test_order_independent(self):
|
|
"""Adding members in different order produces same hash (sorted internally)."""
|
|
from src.core.utils.fileio import compute_archive_content_hash
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname_a = f.name
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname_b = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname_a, "w") as zf:
|
|
zf.writestr("z_last.txt", "last")
|
|
zf.writestr("a_first.txt", "first")
|
|
with zipfile.ZipFile(fname_b, "w") as zf:
|
|
zf.writestr("a_first.txt", "first")
|
|
zf.writestr("z_last.txt", "last")
|
|
assert compute_archive_content_hash(Path(fname_a)) == compute_archive_content_hash(Path(fname_b))
|
|
finally:
|
|
os.unlink(fname_a)
|
|
os.unlink(fname_b)
|
|
|
|
|
|
class TestWriteBackupManifest:
|
|
"""write_backup_manifest — atomically writes .manifest.json sidecar."""
|
|
|
|
def test_writes_manifest(self):
|
|
from src.core.utils.fileio import (
|
|
calculate_sha256,
|
|
compute_archive_content_hash,
|
|
write_backup_manifest,
|
|
read_backup_manifest,
|
|
)
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
# Create a valid ZIP archive
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("test.txt", "content")
|
|
archive_path = Path(fname)
|
|
|
|
manifest_path = write_backup_manifest(archive_path)
|
|
assert manifest_path.exists()
|
|
assert manifest_path.suffix == ".json"
|
|
assert manifest_path.name.endswith(".manifest.json")
|
|
|
|
manifest = read_backup_manifest(manifest_path)
|
|
assert manifest["archive"] == archive_path.name
|
|
assert manifest["archive_sha256"] == calculate_sha256(archive_path)
|
|
assert manifest["content_hash"] == compute_archive_content_hash(archive_path)
|
|
assert manifest["content_hash_algorithm"] == "superset-export-semantic-v1"
|
|
assert manifest["integrity_status"] == "verified"
|
|
assert manifest["size"] == archive_path.stat().st_size
|
|
|
|
manifest_path.unlink()
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_atomic_write(self):
|
|
"""Manifest written atomically (no .tmp file survives)."""
|
|
from src.core.utils.fileio import write_backup_manifest
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("a.txt", "data")
|
|
archive_path = Path(fname)
|
|
manifest_path = write_backup_manifest(archive_path)
|
|
# Scope: only check for this specific manifest's .tmp file
|
|
manifest_tmp = archive_path.with_suffix(".manifest.json.tmp")
|
|
assert not manifest_tmp.exists(), f"Orphan .tmp file left: {manifest_tmp}"
|
|
manifest_path.unlink()
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_extra_metadata(self):
|
|
from src.core.utils.fileio import write_backup_manifest, read_backup_manifest
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("a.txt", "extra")
|
|
manifest_path = write_backup_manifest(
|
|
Path(fname),
|
|
extra={"dashboard_id": 42, "environment": "prod"},
|
|
)
|
|
manifest = read_backup_manifest(manifest_path)
|
|
assert manifest["dashboard_id"] == 42
|
|
assert manifest["environment"] == "prod"
|
|
manifest_path.unlink()
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
|
|
class TestVerifyBackupIntegrity:
|
|
"""verify_backup_integrity — re-computes hashes and compares with manifest."""
|
|
|
|
def test_valid_archive(self):
|
|
from src.core.utils.fileio import verify_backup_integrity, write_backup_manifest
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("data.txt", "integrity check")
|
|
archive_path = Path(fname)
|
|
write_backup_manifest(archive_path)
|
|
|
|
result = verify_backup_integrity(archive_path)
|
|
assert result["status"] == "ok"
|
|
assert result["manifest_sha256"] == result["actual_sha256"]
|
|
assert result["manifest_content_hash"] == result["actual_content_hash"]
|
|
assert result["content_hash_algorithm"] == "superset-export-semantic-v1"
|
|
assert "errors" not in result
|
|
|
|
archive_path.with_suffix(".manifest.json").unlink()
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_missing_manifest(self):
|
|
from src.core.utils.fileio import verify_backup_integrity
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("x.txt", "data")
|
|
result = verify_backup_integrity(Path(fname))
|
|
assert result["status"] == "manifest_missing"
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
def test_tampered_archive(self):
|
|
from src.core.utils.fileio import verify_backup_integrity, write_backup_manifest
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
|
|
fname = f.name
|
|
try:
|
|
with zipfile.ZipFile(fname, "w") as zf:
|
|
zf.writestr("original.txt", "original content")
|
|
archive_path = Path(fname)
|
|
write_backup_manifest(archive_path)
|
|
|
|
# Tamper: modify the archive content
|
|
with zipfile.ZipFile(fname, "a") as zf:
|
|
zf.writestr("tamper.txt", "extra data")
|
|
|
|
result = verify_backup_integrity(archive_path)
|
|
assert result["status"] == "integrity_violated"
|
|
assert "archive_sha256_mismatch" in result["errors"]
|
|
|
|
archive_path.with_suffix(".manifest.json").unlink()
|
|
finally:
|
|
os.unlink(fname)
|
|
|
|
|
|
class TestCreateTempFile:
|
|
"""create_temp_file — context manager for temp resources."""
|
|
|
|
def test_creates_and_cleans_up_file(self):
|
|
from src.core.utils.fileio import create_temp_file
|
|
path = None
|
|
with create_temp_file(content=b"test content", suffix=".zip") as p:
|
|
path = p
|
|
assert path.exists()
|
|
assert path.suffix == ".zip"
|
|
assert path.read_bytes() == b"test content"
|
|
assert not path.exists()
|
|
|
|
def test_no_content(self):
|
|
from src.core.utils.fileio import create_temp_file
|
|
with create_temp_file(suffix=".zip") as p:
|
|
assert p.exists()
|
|
assert p.stat().st_size == 0
|
|
|
|
def test_directory_mode(self):
|
|
from src.core.utils.fileio import create_temp_file
|
|
with create_temp_file(suffix=".dir") as p:
|
|
assert p.is_dir()
|
|
(p / "test.txt").write_text("hello")
|
|
assert (p / "test.txt").exists()
|
|
|
|
def test_dry_run_no_write(self):
|
|
from src.core.utils.fileio import create_temp_file
|
|
with create_temp_file(content=b"data", suffix=".zip", dry_run=True) as p:
|
|
# In dry_run mode, no actual file is created
|
|
pass
|
|
|
|
def test_exception_cleans_up(self):
|
|
from src.core.utils.fileio import create_temp_file
|
|
path = None
|
|
try:
|
|
with create_temp_file(content=b"data", suffix=".txt") as p:
|
|
path = p
|
|
raise ValueError("test error")
|
|
except ValueError:
|
|
pass
|
|
assert path is None or not path.exists()
|
|
|
|
|
|
class TestRemoveEmptyDirectories:
|
|
"""remove_empty_directories — removes empty dirs recursively."""
|
|
|
|
def test_removes_empty_dirs(self):
|
|
from src.core.utils.fileio import remove_empty_directories
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
empty_dir = Path(tmpdir) / "empty"
|
|
empty_dir.mkdir()
|
|
nested_empty = empty_dir / "nested"
|
|
nested_empty.mkdir()
|
|
# Non-empty dir should remain
|
|
non_empty = Path(tmpdir) / "non_empty"
|
|
non_empty.mkdir()
|
|
(non_empty / "file.txt").write_text("data")
|
|
|
|
removed = remove_empty_directories(tmpdir)
|
|
assert removed >= 2 # at least 2 empty dirs removed
|
|
assert non_empty.exists()
|
|
assert not empty_dir.exists()
|
|
|
|
def test_no_empty_dirs(self):
|
|
from src.core.utils.fileio import remove_empty_directories
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
d = Path(tmpdir) / "dir"
|
|
d.mkdir()
|
|
(d / "file.txt").write_text("data")
|
|
removed = remove_empty_directories(tmpdir)
|
|
assert removed == 0
|
|
|
|
|
|
class TestCreateDashboardExport:
|
|
"""create_dashboard_export — packs files into ZIP."""
|
|
|
|
def test_creates_zip_with_files(self):
|
|
from src.core.utils.fileio import create_dashboard_export
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
src = Path(tmpdir) / "src"
|
|
src.mkdir()
|
|
(src / "dashboard.json").write_text('{"key": "value"}')
|
|
(src / "metadata.yaml").write_text("version: 1")
|
|
|
|
zip_path = Path(tmpdir) / "export.zip"
|
|
result = create_dashboard_export(zip_path, [str(src)])
|
|
assert result is True
|
|
assert zip_path.exists()
|
|
|
|
# Verify contents
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
names = zf.namelist()
|
|
assert any("dashboard.json" in n for n in names)
|
|
|
|
def test_excludes_extensions(self):
|
|
from src.core.utils.fileio import create_dashboard_export
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
src = Path(tmpdir) / "src"
|
|
src.mkdir()
|
|
(src / "file.json").write_text("{}")
|
|
(src / "file.log").write_text("log data")
|
|
|
|
zip_path = Path(tmpdir) / "export.zip"
|
|
result = create_dashboard_export(zip_path, [str(src)], exclude_extensions=[".log"])
|
|
assert result is True
|
|
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
names = zf.namelist()
|
|
assert any("file.json" in n for n in names)
|
|
assert not any("file.log" in n for n in names)
|
|
|
|
def test_source_not_found(self):
|
|
from src.core.utils.fileio import create_dashboard_export
|
|
result = create_dashboard_export("/tmp/nonexistent.zip", ["/nonexistent/path"])
|
|
assert result is False
|
|
|
|
|
|
class TestConsolidateArchiveFolders:
|
|
"""consolidate_archive_folders — merges directories with common slug."""
|
|
|
|
def test_consolidates_dirs(self):
|
|
from src.core.utils.fileio import consolidate_archive_folders
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
# Create two dirs with same slug "sales"
|
|
d1 = root / "sales_2024"
|
|
d1.mkdir()
|
|
(d1 / "report1.zip").write_text("zip1")
|
|
d2 = root / "sales_2025"
|
|
d2.mkdir()
|
|
(d2 / "report2.zip").write_text("zip2")
|
|
# Unrelated dir
|
|
d3 = root / "marketing_2024"
|
|
d3.mkdir()
|
|
(d3 / "campaign.zip").write_text("zip3")
|
|
|
|
consolidate_archive_folders(root)
|
|
|
|
# sales_2024 and sales_2025 should be merged into 'sales'
|
|
sales_dir = root / "sales"
|
|
assert sales_dir.is_dir()
|
|
assert (sales_dir / "report1.zip").exists()
|
|
assert (sales_dir / "report2.zip").exists()
|
|
# marketing should remain separate
|
|
assert (root / "marketing_2024").is_dir()
|
|
|
|
def test_single_dir_no_consolidation(self):
|
|
from src.core.utils.fileio import consolidate_archive_folders
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
d = root / "sales_2024"
|
|
d.mkdir()
|
|
(d / "report.zip").write_text("zip")
|
|
consolidate_archive_folders(root)
|
|
assert d.exists()
|
|
|
|
def test_raises_on_invalid_input(self):
|
|
from src.core.utils.fileio import consolidate_archive_folders
|
|
with pytest.raises(AssertionError):
|
|
consolidate_archive_folders("not-a-path")
|
|
|
|
def test_skips_dirs_without_zips(self):
|
|
from src.core.utils.fileio import consolidate_archive_folders
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
(root / "no_zips_here").mkdir()
|
|
consolidate_archive_folders(root)
|
|
assert (root / "no_zips_here").exists()
|
|
# #endregion Test.FileIO.Utils
|