155 lines
6.4 KiB
Python
155 lines
6.4 KiB
Python
"""Recovery policy tests: reclaim only owned scratch, protect recovery/last copies."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'lib'))
|
|
import space
|
|
|
|
|
|
class RemoteIdentityRetry(unittest.TestCase):
|
|
def check(self, mode):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
for file in ('manifest.json', 'COMPLETE'):
|
|
(root/file).write_bytes(b'exact bytes\n')
|
|
remote = root/'rclone'
|
|
remote.write_text('''#!/usr/bin/env python3
|
|
import os, sys, time
|
|
from pathlib import Path
|
|
root = Path(__file__).parent
|
|
file = Path(sys.argv[sys.argv.index('cat')+1]).name
|
|
counter = root/(file+'.attempts')
|
|
attempt = int(counter.read_text())+1 if counter.exists() else 1
|
|
counter.write_text(str(attempt))
|
|
if os.environ['TEST_MODE'] == 'deadline' and attempt == 1:
|
|
time.sleep(20)
|
|
sys.stdout.buffer.write((root/file).read_bytes())
|
|
sys.stdout.buffer.flush()
|
|
if os.environ['TEST_MODE'] == 'late_failure':
|
|
sys.exit(7)
|
|
''')
|
|
remote.chmod(0o700)
|
|
result = subprocess.run(['bash', '-c', '''
|
|
source "$1/lib/common.sh"
|
|
RCLONE_BIN=$2/rclone
|
|
RCLONE_ARGS=()
|
|
RCLONE_CHECK_ATTEMPTS=2
|
|
RCLONE_CHECK_RETRY_DELAY=0
|
|
RCLONE_CHECK_DEADLINE=1
|
|
RCLONE_CHECK_TIMEOUT=1
|
|
RCLONE_CHECK_LOW_LEVEL_RETRIES=1
|
|
remote_identity_check "$2" remote:run
|
|
''', 'test', str(Path(__file__).resolve().parents[1]), str(root)],
|
|
env=dict(os.environ, TEST_MODE=mode), text=True, capture_output=True, timeout=8)
|
|
attempts = {p.name: int(p.read_text()) for p in root.glob('*.attempts')}
|
|
return result, attempts
|
|
|
|
def test_identity_deadline_retries_complete_files(self):
|
|
result, attempts = self.check('deadline')
|
|
self.assertEqual(result.returncode, 0, result.stdout+result.stderr)
|
|
self.assertEqual(attempts, {'manifest.json.attempts': 2, 'COMPLETE.attempts': 2})
|
|
|
|
def test_matching_bytes_with_failed_remote_command_are_not_success(self):
|
|
result, attempts = self.check('late_failure')
|
|
self.assertEqual(result.returncode, 7, result.stdout+result.stderr)
|
|
self.assertEqual(attempts, {'manifest.json.attempts': 2})
|
|
|
|
|
|
@unittest.skipUnless(os.geteuid() == 0, 'root-only cleanup policy')
|
|
class SpaceRecovery(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.work = Path(self.temp.name)
|
|
self.staging = self.work/'staging'
|
|
self.archives = self.work/'archives'
|
|
self.restore = self.work/'restore'
|
|
for p in (self.staging, self.archives, self.restore): p.mkdir(mode=0o700)
|
|
self.run = 'appbackup_2026-09-12_17-07-56_bd1f6193'
|
|
self.stage = self.staging/(self.run+'.5ICERN')
|
|
self.stage.mkdir(mode=0o700)
|
|
(self.stage/'meta').mkdir()
|
|
(self.stage/'meta/run-id.txt').write_text(self.run+'\n')
|
|
(self.stage/'meta/format.txt').write_text('format=2\ncompressor=zstd\n')
|
|
self.partial = self.archives/('.partial-'+self.run)
|
|
self.partial.mkdir(mode=0o700)
|
|
(self.partial/'meta.tar.zst').write_text('incomplete scratch')
|
|
|
|
def clean(self):
|
|
space.clean(self.work, self.staging, self.archives, 'appbackup', self.restore)
|
|
|
|
def test_gemini_failed_compression_reclaims_both_trees(self):
|
|
unknown = self.staging/'keep-manual-files'
|
|
unknown.mkdir(mode=0o700)
|
|
self.clean()
|
|
self.assertFalse(self.stage.exists())
|
|
self.assertFalse(self.partial.exists())
|
|
self.assertTrue(unknown.exists())
|
|
|
|
def test_recovery_journal_blocks_all_cleanup(self):
|
|
(self.work/('recovery-'+self.run+'.json')).write_text('{}')
|
|
with self.assertRaisesRegex(ValueError, 'recovery journal'): self.clean()
|
|
self.assertTrue(self.stage.exists())
|
|
self.assertTrue(self.partial.exists())
|
|
|
|
def test_complete_partial_is_preserved(self):
|
|
(self.partial/'COMPLETE').write_text('possibly only complete snapshot')
|
|
self.clean()
|
|
self.assertTrue(self.stage.exists())
|
|
self.assertTrue(self.partial.exists())
|
|
|
|
def test_nested_symlink_does_not_delete_target(self):
|
|
target = self.work/'production'
|
|
target.mkdir()
|
|
(target/'data').write_text('live')
|
|
(self.stage/'link').symlink_to(target)
|
|
self.clean()
|
|
self.assertEqual((target/'data').read_text(), 'live')
|
|
|
|
def test_mount_is_never_removed(self):
|
|
mount = self.stage/'mounted'
|
|
mount.mkdir()
|
|
with patch('space.os.path.ismount', side_effect=lambda p: p == mount):
|
|
with self.assertRaisesRegex(ValueError, 'mounted'): self.clean()
|
|
self.assertTrue(mount.exists())
|
|
|
|
def test_root_mount_is_never_removed(self):
|
|
with patch('space.os.path.ismount', side_effect=lambda p: p == self.partial):
|
|
with self.assertRaisesRegex(ValueError, 'mounted'): self.clean()
|
|
self.assertTrue(self.partial.exists())
|
|
self.assertTrue(self.stage.exists())
|
|
|
|
def test_file_bind_mount_is_never_removed(self):
|
|
mount = self.partial/'meta.tar.zst'
|
|
with patch('space.os.path.ismount', side_effect=lambda p: p == mount):
|
|
with self.assertRaisesRegex(ValueError, 'mounted'): self.clean()
|
|
self.assertTrue(mount.exists())
|
|
|
|
def test_same_device_bind_mount_from_mountinfo_is_never_removed(self):
|
|
mount = self.partial/'meta.tar.zst'
|
|
with patch('space.mounted_paths', return_value={mount}), patch('space.os.path.ismount', return_value=False):
|
|
with self.assertRaisesRegex(ValueError, 'mounted'): self.clean()
|
|
self.assertTrue(mount.exists())
|
|
|
|
def test_mountinfo_decodes_escaped_mount_paths(self):
|
|
text = '42 1 0:9 / /tmp/backup\\040space/file\\134name rw - ext4 /dev/example rw\n'
|
|
with patch('space.Path.read_text', return_value=text):
|
|
self.assertEqual(space.mounted_paths(), {Path('/tmp/backup space/file\\name')})
|
|
|
|
def test_verification_download_recovered_manual_restore_kept(self):
|
|
verify = self.restore/(self.run+'.abcdef')
|
|
verify.mkdir(mode=0o700)
|
|
(verify/'.verify-only').write_text(self.run+'\n')
|
|
manual = self.restore/(self.run+'.ghijkl')
|
|
manual.mkdir(mode=0o700)
|
|
self.clean()
|
|
self.assertFalse(verify.exists())
|
|
self.assertTrue(manual.exists())
|