Spaces:
Sleeping
Sleeping
File size: 4,942 Bytes
4d10530 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | from __future__ import annotations
import importlib.util
import io
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from types import ModuleType
from unittest.mock import patch
PROJECT_ROOT = Path(__file__).resolve().parent.parent
MODULE_PATH = PROJECT_ROOT / "scripts" / "scratch_cleanup.py"
def _load_module() -> ModuleType:
if not MODULE_PATH.exists():
raise FileNotFoundError(f"Missing cleanup module: {MODULE_PATH}")
spec = importlib.util.spec_from_file_location("scratch_cleanup", MODULE_PATH)
if spec is None or spec.loader is None:
raise AssertionError(f"Unable to load module spec: {MODULE_PATH}")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
class ScratchCleanupTests(unittest.TestCase):
def test_validate_scratch_root_requires_directory_named_scratch(self) -> None:
module = _load_module()
with tempfile.TemporaryDirectory() as temp_dir:
wrong_root = Path(temp_dir) / "not-scratch"
wrong_root.mkdir()
with self.assertRaises(ValueError):
module._validate_scratch_root(wrong_root)
def test_dry_run_collects_targets_without_mutating_filesystem(self) -> None:
module = _load_module()
with tempfile.TemporaryDirectory() as temp_dir:
scratch_root = Path(temp_dir) / "scratch"
nested_dir = scratch_root / "logs"
nested_dir.mkdir(parents=True)
artifact_file = scratch_root / "artifact.png"
artifact_file.write_text("png", encoding="utf-8")
nested_file = nested_dir / "server.log"
nested_file.write_text("log", encoding="utf-8")
result = module.cleanup_scratch(scratch_root, dry_run=True)
self.assertTrue(artifact_file.exists())
self.assertTrue(nested_file.exists())
self.assertEqual(len(result.targets), 2)
self.assertEqual(result.total_files, 2)
self.assertEqual(result.total_directories, 1)
self.assertEqual(result.error_paths, ())
def test_gitkeep_is_not_selected_as_cleanup_target(self) -> None:
module = _load_module()
with tempfile.TemporaryDirectory() as temp_dir:
scratch_root = Path(temp_dir) / "scratch"
scratch_root.mkdir()
keep_file = scratch_root / ".gitkeep"
keep_file.write_text("", encoding="utf-8")
extra_file = scratch_root / "note.txt"
extra_file.write_text("remove me", encoding="utf-8")
result = module.cleanup_scratch(scratch_root, dry_run=True)
target_names = [target.path.name for target in result.targets]
self.assertEqual(target_names, ["note.txt"])
self.assertTrue(keep_file.exists())
def test_apply_recycles_only_top_level_entries(self) -> None:
module = _load_module()
with tempfile.TemporaryDirectory() as temp_dir:
scratch_root = Path(temp_dir) / "scratch"
nested_dir = scratch_root / "nested"
nested_dir.mkdir(parents=True)
top_file = scratch_root / "preview.mp4"
top_file.write_text("video", encoding="utf-8")
nested_file = nested_dir / "inside.txt"
nested_file.write_text("inside", encoding="utf-8")
recycled_paths: list[Path] = []
def _fake_recycle(path: Path) -> None:
recycled_paths.append(path)
with patch.object(module, "_move_to_recycle_bin", side_effect=_fake_recycle):
result = module.cleanup_scratch(scratch_root, dry_run=False)
self.assertEqual(sorted(path.name for path in recycled_paths), ["nested", "preview.mp4"])
self.assertEqual(len(result.targets), 2)
def test_apply_aggregates_recycle_errors_and_returns_non_zero_exit(self) -> None:
module = _load_module()
with tempfile.TemporaryDirectory() as temp_dir:
scratch_root = Path(temp_dir) / "scratch"
scratch_root.mkdir()
(scratch_root / "a.log").write_text("a", encoding="utf-8")
(scratch_root / "b.log").write_text("b", encoding="utf-8")
def _fake_recycle(path: Path) -> None:
if path.name == "a.log":
raise OSError("locked")
stdout_buffer = io.StringIO()
with patch.object(module, "_move_to_recycle_bin", side_effect=_fake_recycle):
with redirect_stdout(stdout_buffer):
exit_code = module.main(["--root", str(scratch_root), "--apply"])
self.assertEqual(exit_code, 1)
output = stdout_buffer.getvalue()
self.assertIn("a.log", output)
self.assertIn("Errors", output)
if __name__ == "__main__":
unittest.main()
|