diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py index 8f82414..e358126 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -75,8 +75,11 @@ def prune_log_files( tracked = _read_log_index(index_path) if tracked is None: tracked = {path.resolve() for path in log_dir.glob("*/logs/*.log")} - tracked.add(current_log_file.resolve()) - candidates = [(path.name, path) for path in tracked if not _same_path(path, current_log_file)] + current_log_file = current_log_file.resolve() + tracked = {path.resolve() for path in tracked} + tracked = {path for path in tracked if path.exists() or path == current_log_file} + tracked.add(current_log_file) + candidates = [(path.name, path) for path in tracked if path != current_log_file] excess_count = len(candidates) + 1 - max_log_files if excess_count > 0: @@ -87,7 +90,7 @@ def prune_log_files( except OSError as exc: logger.warning("Could not prune log file '%s': %s", path, exc) - tracked = {path for path in tracked if path.exists() or _same_path(path, current_log_file)} + tracked = {path for path in tracked if path.exists() or path == current_log_file} try: write_private_json(index_path, {"version": 1, "logs": sorted(str(path) for path in tracked)}) except (OSError, TypeError, ValueError) as exc: @@ -121,10 +124,3 @@ def _runtime_directory_error(path: Path, cache_root: Path, exc: OSError) -> str: f"Check permissions on that directory. If the Base cache root '{cache_root}' is unusable, " "set BASE_CACHE_DIR to a writable directory." ) - - -def _same_path(left: Path, right: Path) -> bool: - try: - return left.resolve() == right.resolve() - except OSError: - return left.absolute() == right.absolute() diff --git a/tests/test_app_log_retention.py b/tests/test_app_log_retention.py index 9d24228..aee4ac4 100644 --- a/tests/test_app_log_retention.py +++ b/tests/test_app_log_retention.py @@ -1,6 +1,8 @@ from __future__ import annotations import importlib.util +import json +import logging import os import tempfile import unittest @@ -8,6 +10,7 @@ from unittest import mock import base_cli +from base_cli._runtime import prune_log_files from base_cli.testing import invoke @@ -18,6 +21,39 @@ def write_log_file(path: Path, mtime: int) -> None: class AppLogRetentionTests(unittest.TestCase): + def test_ignores_stale_paths_in_retention_index(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + log_dir = Path(tmpdir) / "runs" / "seed" / "logs" + old_a = log_dir / "20260620T120000_a.log" + old_b = log_dir / "20260621T120000_b.log" + old_c = log_dir / "20260622T120000_c.log" + current = log_dir / "20260623T120000_current.log" + phantom = log_dir / "deleted" / "phantom.log" + for path, mtime in ((old_a, 1), (old_b, 2), (old_c, 3)): + write_log_file(path, mtime) + (log_dir / ".base-cli-log-index.json").write_text( + json.dumps( + { + "version": 1, + "logs": [str(path.resolve()) for path in (old_a, old_b, old_c, phantom)], + } + ), + encoding="utf-8", + ) + + prune_log_files(log_dir, current, 3, logging.getLogger(__name__)) + + self.assertFalse(old_a.exists()) + self.assertTrue(old_b.exists()) + self.assertTrue(old_c.exists()) + current.write_text("current log\n", encoding="utf-8") + self.assertEqual(len(tuple(log_dir.rglob("*.log"))), 3) + index = json.loads((log_dir / ".base-cli-log-index.json").read_text(encoding="utf-8")) + self.assertEqual( + {Path(path) for path in index["logs"]}, + {old_b.resolve(), old_c.resolve(), current.resolve()}, + ) + @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") def test_uses_retention_index_after_initial_discovery(self) -> None: app = base_cli.App(name="retention-index", max_log_files=2)