191 lines
9.1 KiB
Python
191 lines
9.1 KiB
Python
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))
|