import json from pathlib import Path import shutil import tempfile import unittest from unittest.mock import patch from test_updater import u class RetentionRunner: def __init__(self, log, owner): self.log, self.owner = log, owner def run(self, command): if command[:2] == ['docker', 'ps']: return 'container' if self.owner.mounts else '' if command[:2] == ['docker', 'inspect']: return json.dumps([{'Mounts': self.owner.mounts}]) raise AssertionError(command) class RetentionTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) (self.root / 'docker-compose.yml').write_text('unchanged compose') self.mounts = [] self.updater = u.Updater(self.root, runner_factory=lambda log: RetentionRunner(log, self)) self.jobs = [self.backup(i) for i in range(1, 5)] latest = u.read_json(self.jobs[-1] / 'status.json') u.write_json(self.updater.home / 'installed.json', { 'manifest': {'version': latest['version'], 'commit': latest['revision']}, 'composeSha256': u.file_sha(self.root / 'docker-compose.yml')}) u.write_json(self.updater.home / 'status.json', latest) def backup(self, index, *, name=None, status='succeeded', command='apply'): job = self.updater.home / (name or f'202601{index:02d}-120000-{index:08x}') directory = job / 'backup' directory.mkdir(parents=True) files = ['database.dump', 'globals.sql', 'files.tar', 'files.list', 'containers.json', 'config/.env', 'config/docker-compose.yml'] for name in files: path = directory / name path.parent.mkdir(exist_ok=True) path.write_text(f'{index}: {name}') u.write_json(directory / 'complete.json', { 'database': 'database.dump', 'files': 'files.tar', 'storageProvider': 'minio', 'sha256': {name: u.file_sha(directory / name) for name in files}}) u.write_json(job / 'status.json', { 'command': command, 'status': status, 'phase': 'complete', 'job': str(job), 'backup': str(directory), 'maintenance': False, 'recoveryRequired': False, 'version': f'3.{index}.0', 'revision': str(index) * 40, 'startedAt': f'2026-01-{index:02d}T12:00:00+00:00'}) (job / 'commands.log').write_text('audit log stays') return job def test_keep_two_and_preserve_failed_migration_and_audit_records(self): special = [self.backup(5, status='failed'), self.backup(6, name='storage-migration'), self.backup(7, name='foreign'), self.backup(8, command='check')] report = self.updater.prune_backups() self.assertEqual(report['status'], 'complete') self.assertEqual(len(report['removed']), 2) self.assertEqual(len(report['retained']), 2) for job in self.jobs[:2]: self.assertFalse((job / 'backup').exists()) self.assertEqual(u.read_json(job / 'backup-pruned.json')['status'], 'removed') self.assertTrue((job / 'commands.log').is_file()) for job in self.jobs[2:] + special: self.assertTrue((job / 'backup/complete.json').is_file()) self.assertEqual(self.updater.prune_backups()['removed'], []) def test_dry_run_checks_but_does_not_delete_or_write_markers(self): result = self.updater.prune_backups(dry_run=True) self.assertEqual(len(result['planned']), 2) self.assertFalse((self.updater.home / 'backup-retention.json').exists()) for job in self.jobs: self.assertTrue((job / 'backup').is_dir()) self.assertFalse((job / 'backup-pruned.json').exists()) def test_corrupt_either_retained_generation_prevents_all_deletions(self): for job in self.jobs[2:]: with self.subTest(job=job.name): path = job / 'backup/database.dump' original = path.read_bytes() path.write_bytes(b'corrupt') with self.assertRaisesRegex(u.UpdateError, 'Prüfsumme'): self.updater.prune_backups() self.assertTrue(all((p / 'backup').exists() for p in self.jobs)) path.write_bytes(original) def test_incomplete_retained_backup_prevents_deletion(self): (self.jobs[-1] / 'backup/globals.sql').unlink() with self.assertRaisesRegex(u.UpdateError, 'fehlt'): self.updater.prune_backups() self.assertTrue((self.jobs[0] / 'backup').exists()) def test_seaweedfs_requires_image_and_provider_snapshot(self): path = self.jobs[-1] / 'backup/complete.json' manifest = u.read_json(path) manifest['storageProvider'] = 'seaweedfs' u.write_json(path, manifest) with self.assertRaisesRegex(u.UpdateError, 'Unvollständiges'): self.updater.prune_backups() self.assertTrue((self.jobs[0] / 'backup').exists()) def test_unknown_files_and_links_are_preserved(self): (self.jobs[0] / 'backup/foreign.txt').write_text('do not delete') outside = self.root / 'outside' outside.write_text('keep') (self.jobs[1] / 'backup/files.tar').unlink() (self.jobs[1] / 'backup/files.tar').symlink_to(outside) result = self.updater.prune_backups() self.assertEqual(len(result['skipped']), 2) self.assertEqual(result['removed'], []) self.assertEqual(outside.read_text(), 'keep') def test_symlinked_backup_directory_is_not_followed(self): backup = self.jobs[0] / 'backup' outside = self.root / 'outside' backup.rename(outside) backup.symlink_to(outside, target_is_directory=True) result = self.updater.prune_backups() self.assertTrue(outside.is_dir()) self.assertTrue(backup.is_symlink()) self.assertEqual(len(result['skipped']), 1) def test_manifest_path_escape_is_rejected(self): path = self.jobs[-1] / 'backup/complete.json' manifest = u.read_json(path) manifest['sha256']['../outside'] = 'a' * 64 u.write_json(path, manifest) with self.assertRaisesRegex(u.UpdateError, 'Ungültiger Pfad'): self.updater.prune_backups() self.assertTrue((self.jobs[0] / 'backup').exists()) def test_container_mount_protects_backup_but_metrics_root_mount_does_not(self): self.mounts = [{'Source': '/', 'RW': False}, {'Source': str(self.jobs[0] / 'backup/files.tar'), 'RW': False}] result = self.updater.prune_backups() self.assertEqual(len(result['skipped']), 1) self.assertTrue((self.jobs[0] / 'backup').exists()) self.assertFalse((self.jobs[1] / 'backup').exists()) def test_same_device_bind_mount_is_protected(self): read_text = Path.read_text mount = self.jobs[0] / 'backup/config' def read(path, *args, **kwargs): if path == Path('/proc/self/mountinfo'): return f'1 2 3:4 / {mount} rw - ext4 /dev/test rw\n' return read_text(path, *args, **kwargs) with patch.object(Path, 'read_text', read): result = self.updater.prune_backups() self.assertEqual(len(result['skipped']), 1) self.assertTrue((self.jobs[0] / 'backup').exists()) def test_partial_deletion_can_resume_without_touching_retained_backups(self): original = shutil.rmtree def fail(path, *args, **kwargs): if path == self.jobs[0] / 'backup': (path / 'complete.json').unlink() (path / 'database.dump').unlink() raise OSError('simulated disk error') return original(path, *args, **kwargs) with patch.object(u.shutil, 'rmtree', fail): result = self.updater.prune_backups() self.assertEqual(len(result['skipped']), 1) self.assertEqual(u.read_json(self.jobs[0] / 'backup-pruned.json')['status'], 'deleting') result = self.updater.prune_backups() self.assertEqual(len(result['removed']), 1) self.assertTrue(all((p / 'backup').exists() for p in self.jobs[2:])) def test_recovery_running_and_lock_block_cleanup(self): for state in ({'recoveryRequired': True}, {'maintenance': True}, {'status': 'running'}, {'phase': 'storage_migration'}): with self.subTest(state=state): u.write_json(self.updater.home / 'status.json', state) with self.assertRaises(u.UpdateError): self.updater.prune_backups() u.write_json(self.updater.home / 'status.json', {'status': 'failed', 'recoveryRequired': False}) with self.updater.lock(), self.assertRaises(u.UpdateError): self.updater.prune_backups() self.assertEqual(len(self.updater.prune_backups()['removed']), 2) def test_changed_installation_blocks_cleanup(self): (self.root / 'docker-compose.yml').write_text('manually restored') with self.assertRaisesRegex(u.UpdateError, 'installierten Stand'): self.updater.prune_backups() self.assertTrue(all((p / 'backup').exists() for p in self.jobs))