#!/usr/bin/env python3 """Exercise destructive cleanup only in small, disposable Git repositories.""" import fcntl import contextlib import importlib.machinery import importlib.util import io import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from unittest.mock import patch PROJECT = Path(__file__).resolve().parents[2] SCRIPT = PROJECT / 'tools/clean-builds' loader = importlib.machinery.SourceFileLoader('clean_builds', str(SCRIPT)) spec = importlib.util.spec_from_loader(loader.name, loader) clean = importlib.util.module_from_spec(spec) loader.exec_module(clean) class CleanupTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(prefix='fds-clean-test.') self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) / 'project' (self.root / 'tools').mkdir(parents=True) (self.root / 'out/manifests').mkdir(parents=True) shutil.copyfile(SCRIPT, self.root / 'tools/clean-builds') subprocess.run(['git', 'init', '-q', self.root], check=True) def file(self, name, content='fixture'): path = self.root / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) return path def run_cli(self, *args): return subprocess.run([sys.executable, self.root / 'tools/clean-builds', *args], cwd=self.root, text=True, capture_output=True, timeout=30) def selected(self): return {p.relative_to(self.root).as_posix() for p in clean.candidates(self.root)[0]} def test_cleanup_and_repeat_preserve_published_and_user_files(self): old = self.file('out/rootfs-build.OLD123/root/data') self.file('target/debug/program') keep = [ self.file('out/rootfs-build.NEW123/rootfs.tar'), self.file('out/fds-os-0.1.0/release.img'), self.file('out/inputs-m12-v5/lock.json'), self.file('out/rebuild-m12-v5-a/out/image.img'), self.file('out/my-emulator/data.qcow2'), self.file('out/emulator/session.json'), self.file('out/my-cartridge.img'), self.file('out/cache/rootfs/package.xbps'), self.file('out/logs/build.log'), self.file('out/manifests/acceptance.json'), self.file('out/.gitkeep'), self.file('.host/xbps/tool'), self.file('.host/void-packages/hostdir/download'), self.file('vendor/void-packages/xbps-src'), ] (self.root / 'out/rootfs-cli.tar').symlink_to('rootfs-build.NEW123/rootfs.tar') preview = self.run_cli('--dry-run') self.assertEqual(preview.returncode, 0, preview.stderr) self.assertTrue(old.exists()) self.assertFalse((self.root / 'out/.clean.lock').exists()) result = self.run_cli() self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(old.exists()) self.assertFalse((self.root / 'target').exists()) self.assertTrue(all(p.read_text() == 'fixture' for p in keep)) self.assertEqual((self.root / 'out/rootfs-cli.tar').read_text(), 'fixture') repeat = self.run_cli() self.assertEqual(repeat.returncode, 0, repeat.stderr) self.assertIn('removed 0 disposable directories', repeat.stdout) def test_text_pointers_latest_fixture_markers_and_transitive_links(self): self.file('out/emu-test.old000/normal/data.qcow2') self.file('out/emu-test.new000/normal/data.qcow2') self.file('out/workstation-emulator-current.txt', str(self.root / 'out/emu-test.new000') + '\n') self.file('out/dasung-s6.old000/compiled/a') self.file('out/dasung-s6.new000/compiled/a') self.file('out/manifests/dasung-s6-database.txt', str(self.root / 'out/dasung-s6.new000/compiled')) self.file('out/workstation-images.new000/acceptance.json') self.file('out/workstation-images-current.txt', 'out/workstation-images.new000\n') self.file('out/m9-images.old000/root/program.img') self.file('out/m9-images.new000/root/program.img') os.utime(self.root / 'out/m9-images.old000', ns=(1, 1)) self.file('out/m8-vm.kept00/.fds-keep', '') self.file('out/m8-vm.other0/disk.img') (self.root / 'out/m8-vm.kept00/dependency').symlink_to('../m8-vm.other0') self.assertEqual(self.selected(), { 'out/emu-test.old000', 'out/dasung-s6.old000', 'out/m9-images.old000'}) def test_symlinks_do_not_delete_external_content(self): outside = Path(self.temporary.name) / 'outside' outside.mkdir() (outside / 'data').write_text('precious') (self.root / 'target').symlink_to(outside) (self.root / 'out/m8-vm.abcdef').symlink_to(outside) work = self.file('out/m8-vm.ghijkl/data').parent (work / 'external').symlink_to(outside) result = self.run_cli() self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(work.exists()) self.assertTrue((self.root / 'target').is_symlink()) self.assertEqual((outside / 'data').read_text(), 'precious') shutil.rmtree(self.root / 'out') (self.root / 'out').symlink_to(outside) self.assertNotEqual(self.run_cli().returncode, 0) self.assertEqual((outside / 'data').read_text(), 'precious') def test_tracked_files_are_kept(self): self.file('out/m2-vm.abcdef/notes') subprocess.run(['git', '-C', self.root, 'add', 'out/m2-vm.abcdef/notes'], check=True) self.assertEqual(self.selected(), set()) def test_directory_permissions_and_hardlinks(self): kept = self.file('out/kept-file') kept.chmod(0o444) root = self.root / 'out/rootfs-build.abcdef/root' root.mkdir(parents=True) os.link(kept, root / 'file') root.chmod(0o555) result = self.run_cli() self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(root.exists()) self.assertEqual(kept.stat().st_mode & 0o777, 0o444) self.assertEqual(kept.read_text(), 'fixture') def test_held_build_lock_rejects_without_deletion(self): old = self.file('out/m2-vm.abcdef/image') with (self.root / 'out/.rootfs.lock').open('w') as lock: fcntl.flock(lock, fcntl.LOCK_EX) result = self.run_cli() self.assertNotEqual(result.returncode, 0) self.assertIn('locked', result.stderr) self.assertTrue(old.exists()) def test_live_process_rejects_without_deletion(self): old = self.file('out/m2-vm.abcdef/image') child = subprocess.Popen( [sys.executable, '-c', 'import sys; print("ready", flush=True); sys.stdin.read()'], cwd=old.parent, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) try: self.assertEqual(child.stdout.readline().strip(), 'ready') result = self.run_cli() self.assertNotEqual(result.returncode, 0) self.assertIn('is using', result.stderr) self.assertTrue(old.exists()) finally: child.communicate(timeout=10) def test_mount_rejection_including_escaped_spaces(self): mount = self.root / 'out/m2-vm.abcdef/a directory' line = '1 0 0:1 / ' + str(mount).replace(' ', r'\040') + ' rw - tmpfs tmpfs rw\n' with self.assertRaisesRegex(ValueError, 'mounted path'): clean.check_mounts([mount.parent], line) clean.check_mounts([self.root / 'target'], line) def test_newly_published_output_aborts_before_any_deletion(self): old = self.file('out/system-build.abcdef/system.img') def publish(_paths): (self.root / 'out/fds-system-cli.img').symlink_to('system-build.abcdef/system.img') return 0 with patch.object(clean, 'footprint', side_effect=publish), contextlib.redirect_stdout(io.StringIO()): with self.assertRaisesRegex(ValueError, 'selection changed'): clean.clean(self.root, False) self.assertTrue(old.exists()) if __name__ == '__main__': unittest.main(verbosity=2)