busya
54a0317e98
- executors: 13 tests, 100% coverage (init/shutdown/run_blocking/run_cpu_blocking) - fileio_utils: 33 tests, 45% coverage (sanitize_filename, get_filename_from_headers, calculate_crc32, create_temp_file, remove_empty_directories, create_dashboard_export, consolidate_archive_folders) - client_registry: 14 tests, 92% coverage (get_client, get_superset_client, get_semaphore, get_auth_lock, shutdown) - cot_logger: 24 tests, 100% coverage (seed/set/get trace_id, push/pop span, structured log with markers, MarkerLogger proxy) - encryption: 12 tests, 100% coverage (key validation, encrypt/decrypt cycle) - rate_limiter: 9 tests, 100% coverage (ban logic, window pruning, per-IP isolation) - auth/logger: 10 tests, 100% coverage (_mask_details, log_security_event)
311 lines
12 KiB
Python
311 lines
12 KiB
Python
# #region Test.FileIO.Utils [C:3] [TYPE Module] [SEMANTICS test,fileio,utils,filename,sanitize,crc32]
|
|
# @BRIEF Tests for core/utils/fileio.py — pure utility functions: sanitize_filename, get_filename_from_headers, calculate_crc32, 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 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
|