# #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" 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