180 lines
8.0 KiB
Python
180 lines
8.0 KiB
Python
#!/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('vendor/void-packages/hostdir/download'),
|
|
]
|
|
(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)
|