From 0cedc26057b9ed859de39090db036cd52528089b Mon Sep 17 00:00:00 2001 From: Luis Raimundo Date: Sat, 6 Jun 2026 17:03:45 +0100 Subject: [PATCH] Add focused output path tests --- tests/test_output_paths.py | 270 +++++++++++++++++++++++++++++++++++++ 1 file changed, 270 insertions(+) create mode 100644 tests/test_output_paths.py diff --git a/tests/test_output_paths.py b/tests/test_output_paths.py new file mode 100644 index 0000000..e655018 --- /dev/null +++ b/tests/test_output_paths.py @@ -0,0 +1,270 @@ +"""Focused pytest coverage for export path helpers and stale-export cleanup.""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from types import SimpleNamespace +from typing import Iterator + +import pytest + +import registral_dispersion.output_paths as output_paths +from registral_dispersion.output_paths import ( + cleanup_stale_exports, + export_directory, + new_export_path, +) + + +# --------------------------------------------------------------------------- +# Fake Path-like helpers for OSError branches +# --------------------------------------------------------------------------- + + +class _FakeExportFile: + def __init__( + self, + name: str, + mtime: float, + *, + stat_raises: bool = False, + unlink_raises: bool = False, + ) -> None: + self.name = name + self._mtime = mtime + self.stat_raises = stat_raises + self.unlink_raises = unlink_raises + self.unlink_calls = 0 + + def is_file(self) -> bool: + return True + + def stat(self): + if self.stat_raises: + raise OSError(f"stat failed for {self.name}") + return SimpleNamespace(st_mtime=self._mtime) + + def unlink(self) -> None: + self.unlink_calls += 1 + if self.unlink_raises: + raise OSError(f"unlink failed for {self.name}") + + +class _FakeExportDir: + def __init__(self, files: list[_FakeExportFile] | None = None) -> None: + self._files = list(files or []) + self.iterdir_calls = 0 + + def iterdir(self) -> Iterator[_FakeExportFile]: + self.iterdir_calls += 1 + return iter(self._files) + + +class _FakeExportDirSecondIterdirFails: + def __init__(self, files: list[_FakeExportFile] | None = None) -> None: + self._files = list(files or []) + self.iterdir_calls = 0 + + def iterdir(self) -> Iterator[_FakeExportFile]: + self.iterdir_calls += 1 + if self.iterdir_calls == 1: + return iter(self._files) + raise OSError("directory listing failed") + + +class _FakeExportDirIterdirAlwaysFails: + def iterdir(self) -> Iterator[_FakeExportFile]: + raise OSError("directory access denied") + yield # pragma: no cover + + +# --------------------------------------------------------------------------- +# cleanup_stale_exports — safe return on directory errors +# --------------------------------------------------------------------------- + + +def test_cleanup_returns_safely_when_directory_iterdir_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + output_paths, + "export_directory", + lambda: _FakeExportDirIterdirAlwaysFails(), + ) + cleanup_stale_exports(max_age_seconds=3600, max_files=10) + + +def test_cleanup_returns_safely_when_second_iterdir_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = time.time() + old = _FakeExportFile("old.csv", now - 10_000) + monkeypatch.setattr( + output_paths, + "export_directory", + lambda: _FakeExportDirSecondIterdirFails([old]), + ) + cleanup_stale_exports(max_age_seconds=3600, max_files=10) + + +# --------------------------------------------------------------------------- +# cleanup_stale_exports — age-based deletion +# --------------------------------------------------------------------------- + + +def test_cleanup_removes_old_files_and_keeps_recent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("REGISTRAL_DISPERSION_CACHE_DIR", str(tmp_path)) + old_file = tmp_path / "old.csv" + recent_file = tmp_path / "recent.csv" + old_file.write_text("old", encoding="utf-8") + recent_file.write_text("recent", encoding="utf-8") + + now = time.time() + os.utime(old_file, (now - 100_000, now - 100_000)) + os.utime(recent_file, (now, now)) + + cleanup_stale_exports(max_age_seconds=3600, max_files=400) + + assert not old_file.exists() + assert recent_file.exists() + + +# --------------------------------------------------------------------------- +# cleanup_stale_exports — per-file stat/unlink OSError +# --------------------------------------------------------------------------- + + +def test_cleanup_continues_when_individual_stat_or_unlink_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = time.time() + bad_stat = _FakeExportFile("bad-stat.csv", now, stat_raises=True) + bad_unlink = _FakeExportFile("bad-unlink.csv", now - 10_000, unlink_raises=True) + monkeypatch.setattr( + output_paths, + "export_directory", + lambda: _FakeExportDir([bad_stat, bad_unlink]), + ) + + cleanup_stale_exports(max_age_seconds=3600, max_files=10) + + assert bad_unlink.unlink_calls == 1 + + +# --------------------------------------------------------------------------- +# cleanup_stale_exports — max_files pruning +# --------------------------------------------------------------------------- + + +def test_cleanup_prunes_to_max_files_by_mtime( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("REGISTRAL_DISPERSION_CACHE_DIR", str(tmp_path)) + now = time.time() + paths: list[Path] = [] + for i in range(5): + path = tmp_path / f"export_{i}.csv" + path.write_text(str(i), encoding="utf-8") + mtime = now - (5 - i) * 100 + os.utime(path, (mtime, mtime)) + paths.append(path) + + cleanup_stale_exports(max_age_seconds=1_000_000, max_files=2) + + assert not paths[0].exists() + assert not paths[1].exists() + assert not paths[2].exists() + assert paths[3].exists() + assert paths[4].exists() + + +def test_cleanup_max_files_pruning_survives_unlink_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = time.time() + oldest = _FakeExportFile("oldest.csv", now - 400, unlink_raises=True) + middle = _FakeExportFile("middle.csv", now - 300) + newer = _FakeExportFile("newer.csv", now - 200) + newest = _FakeExportFile("newest.csv", now) + monkeypatch.setattr( + output_paths, + "export_directory", + lambda: _FakeExportDir([oldest, middle, newer, newest]), + ) + + cleanup_stale_exports(max_age_seconds=1_000_000, max_files=2) + + assert oldest.unlink_calls == 1 + assert middle.unlink_calls == 1 + assert newer.unlink_calls == 0 + assert newest.unlink_calls == 0 + + +# --------------------------------------------------------------------------- +# export_directory / new_export_path +# --------------------------------------------------------------------------- + + +def test_export_directory_creates_missing_parent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + export_dir = tmp_path / "nested" / "exports" + monkeypatch.setenv("REGISTRAL_DISPERSION_CACHE_DIR", str(export_dir)) + result = export_directory() + assert result == export_dir + assert export_dir.is_dir() + + +@pytest.mark.parametrize( + "env_var", + [ + "REGISTRAL_DISPERSION_CACHE_DIR", + "REGISTER_UNIFORMITY_CACHE_DIR", + "HOMOGENEITY_CACHE_DIR", + ], +) +def test_export_directory_respects_env_aliases( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, env_var: str +) -> None: + for key in ( + "REGISTRAL_DISPERSION_CACHE_DIR", + "REGISTER_UNIFORMITY_CACHE_DIR", + "HOMOGENEITY_CACHE_DIR", + ): + monkeypatch.delenv(key, raising=False) + monkeypatch.setenv(env_var, str(tmp_path)) + assert export_directory() == tmp_path + + +def test_new_export_path_uses_prefix_and_suffix( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("REGISTRAL_DISPERSION_CACHE_DIR", str(tmp_path)) + path = Path(new_export_path("dispersion_", ".csv")) + assert path.parent == tmp_path + assert path.name.startswith("dispersion_") + assert path.suffix == ".csv" + + +def test_new_export_path_supports_empty_suffix( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("REGISTRAL_DISPERSION_CACHE_DIR", str(tmp_path)) + path = Path(new_export_path("data_", "")) + assert path.parent == tmp_path + assert path.name.startswith("data_") + + +def test_new_export_path_preserves_spaces_and_dots_in_prefix( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("REGISTRAL_DISPERSION_CACHE_DIR", str(tmp_path)) + prefix = "my export.v2." + path = Path(new_export_path(prefix, ".png")) + assert path.parent == tmp_path + assert path.name.startswith(prefix) + assert path.name.endswith(".png")