"""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())