import importlib.util import json import os from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('runner', ROOT / 'scripts/nextcloud-borg-runner.py') m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) class BackupTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.env = patch.dict(os.environ, { 'STATE_DIR': str(self.root/'state'), 'REPO_DIR': str(self.root/'repo'), 'STAGING_DIR': str(self.root/'stage'), 'REMOTE': ':local:'+str(self.root/'source'), 'ARCHIVE_PREFIX': 'nextcloud', 'BORG_COMPRESSION': 'zstd,1', 'EXCLUDE_FILE': str(self.root/'excludes'), 'LOCK_FILE': str(self.root/'state/lock'), 'MIN_FREE_STAGE_GB': '0', 'MIN_FREE_REPO_GB': '0', 'MAX_STAGE_DELETES': '1000', 'PRUNE_AFTER_BACKUP': 'false', 'BORG_PASSPHRASE': '', 'BORG_PASSCOMMAND': '', 'BORG_NON_INTERACTIVE': '1', }) self.env.start() self.runner = m.Runner() def tearDown(self): self.env.stop() self.tmp.cleanup() def manifest(self, count=10, size=100): return {f'f{i}': {'size': size, 'mtime': 'now', 'hashes': {}} for i in range(count)} def test_empty_and_shrinking_sources_rejected(self): for manifest in ({}, self.manifest(7), self.manifest(10, 70)): with self.subTest(manifest=manifest), self.assertRaises(m.BackupError): self.runner.guard_source(manifest, {'files': 10, 'bytes': 1000}) self.runner.guard_source(self.manifest(8), {'files': 10, 'bytes': 1000}) def test_wrong_nas_rejected_before_writes(self): self.runner.repo.mkdir() with patch.object(self.runner, 'run', side_effect=['', json.dumps({'filesystems': [ {'source': '/dev/sda1', 'fstype': 'ext4'}]})]): with self.assertRaisesRegex(m.BackupError, 'Wrong NAS'): self.runner.nas() self.assertEqual(list(self.runner.repo.iterdir()), []) def test_stacked_autofs_uses_actual_backing_device(self): self.runner.repo.mkdir() device = self.runner.repo.stat().st_dev dev = f'{os.major(device)}:{os.minor(device)}' with patch.object(self.runner, 'run', side_effect=['', json.dumps({'filesystems': [ {'source': '-hosts', 'fstype': 'autofs', 'maj:min': '0:99999'}, {'source': 'thor:/volume1/NetBackup', 'fstype': 'nfs4', 'maj:min': dev}]})]): self.runner.nas() def test_symlink_and_unowned_stage_rejected(self): (self.root/'source').mkdir() (self.root/'stage').symlink_to(self.root/'source', target_is_directory=True) with self.assertRaisesRegex(m.BackupError, 'symlink'): self.runner.staging_preflight(self.manifest()) (self.root/'stage').unlink() (self.root/'stage').mkdir() (self.root/'stage/file').write_text('private') with self.assertRaisesRegex(m.BackupError, 'unowned'): self.runner.staging_preflight(self.manifest()) def test_dedicated_stage_mount_missing_or_wrong_uuid_does_not_create_directory(self): disk = self.root/'disk' self.runner.stage = disk/'private/staging' for actual in ({'target': '/', 'uuid': 'system-disk', 'fstype': 'btrfs'}, {'target': str(disk), 'uuid': 'wrong-disk', 'fstype': 'btrfs'}): with self.subTest(actual=actual), patch.dict(os.environ, { 'EXPECTED_STAGE_MOUNT': str(disk), 'EXPECTED_STAGE_UUID': 'archive-disk'}), \ patch.object(self.runner, 'run', return_value=json.dumps({'filesystems': [actual]})): with self.assertRaisesRegex(m.BackupError, 'missing or has the wrong identity'): self.runner.staging_preflight(self.manifest()) self.assertFalse(disk.exists()) def test_dedicated_stage_mount_verified_before_owned_stage_creation(self): disk = self.root/'disk' disk.mkdir() self.runner.stage = disk/'staging' with patch.dict(os.environ, {'EXPECTED_STAGE_MOUNT': str(disk), 'EXPECTED_STAGE_UUID': 'archive-disk'}), \ patch.object(self.runner, 'run', side_effect=[json.dumps({'filesystems': [ {'target': str(disk), 'uuid': 'archive-disk', 'fstype': 'btrfs'}]}), json.dumps({'filesystems': [{'fstype': 'btrfs'}]})]): self.runner.staging_preflight(self.manifest()) self.assertEqual(m.read_json(disk/'staging.owner.json'), {'remote': self.runner.remote, 'stage': str(self.runner.stage)}) def test_dedicated_stage_mount_configuration_must_be_paired_and_contain_stage(self): for values in ({'EXPECTED_STAGE_MOUNT': str(self.root), 'EXPECTED_STAGE_UUID': ''}, {'EXPECTED_STAGE_MOUNT': '', 'EXPECTED_STAGE_UUID': 'archive-disk'}, {'EXPECTED_STAGE_MOUNT': str(self.root/'other-disk'), 'EXPECTED_STAGE_UUID': 'archive-disk'}): with self.subTest(values=values), patch.dict(os.environ, values): with self.assertRaises(m.BackupError): self.runner.staging_preflight(self.manifest()) self.assertFalse(self.runner.stage.exists()) def test_subprocess_failure_never_becomes_success(self): with patch.object(self.runner, 'perform', side_effect=m.BackupError('network failure')), \ patch.object(self.runner, 'notify') as notify: with self.assertRaises(m.BackupError): self.runner.backup() notify.assert_called_once() self.assertFalse(self.runner.success_file.exists()) self.assertEqual(m.read_json(self.runner.attempt_file)['status'], 'failed') def test_due_skip_and_clock_rollback(self): import time self.runner.success = {'completed_at': time.time()} with patch.object(self.runner, 'perform') as perform: self.runner.backup(True) perform.assert_not_called() self.runner.success = {'completed_at': time.time()+86400} with patch.object(self.runner, 'perform') as perform: self.runner.backup(True) perform.assert_called_once() def test_changed_cloud_never_reaches_borg_create(self): calls = [] def run(args, **kwargs): calls.append(list(map(str, args))) return json.dumps({'repository': {'id': 'test'}}) with patch.object(self.runner, 'nas'), patch.object(self.runner, 'run', side_effect=run), \ patch.object(self.runner, 'baseline', return_value={}), \ patch.object(self.runner, 'manifest', side_effect=[self.manifest(), self.manifest(11)]), \ patch.object(self.runner, 'staging_preflight'), patch.object(self.runner, 'cloud'), patch.object(self.runner, 'verify_local_hashes'): with self.assertRaisesRegex(m.BackupError, 'Cloud changed'): self.runner.perform() self.assertFalse(any('create' in x for x in calls)) self.assertFalse(self.runner.success_file.exists()) def test_prune_failure_retried_even_when_backup_current(self): import time self.runner.success = {'completed_at': time.time()} m.atomic_json(self.runner.attempt_file, {'phase': 'retention', 'status': 'failed'}) with patch.object(self.runner, 'nas'), patch.object(self.runner, 'maintenance') as maintenance: self.runner.backup(True) maintenance.assert_called_once() self.assertEqual(m.read_json(self.runner.attempt_file)['status'], 'success') def test_repository_replacement_rejected(self): m.atomic_json(self.runner.state/'repository.json', {'repo': str(self.runner.repo), 'id': 'old'}) with patch.object(self.runner, 'nas'), patch.object(self.runner, 'run', return_value=json.dumps( {'repository': {'id': 'new'}})): with self.assertRaisesRegex(m.BackupError, 'identity changed'): self.runner.perform() def test_failed_sync_or_check_never_creates_or_prunes(self): for failing in ('sync', 'check'): calls = [] def cloud(command, *args, **kwargs): if command == failing: raise m.BackupError('simulated disconnect') with self.subTest(command=failing), patch.object(self.runner, 'nas'), \ patch.object(self.runner, 'run', side_effect=lambda args, **kw: ( calls.append(list(map(str, args))) or json.dumps({'repository': {'id': 'test'}}))), \ patch.object(self.runner, 'baseline', return_value={}), \ patch.object(self.runner, 'manifest', return_value=self.manifest()), \ patch.object(self.runner, 'staging_preflight'), \ patch.object(self.runner, 'cloud', side_effect=cloud): with self.assertRaises(m.BackupError): self.runner.perform() self.assertFalse(any('create' in x or 'prune' in x for x in calls)) def test_timeout_stops_child(self): import sys with self.assertRaises(subprocess.TimeoutExpired): self.runner.run([sys.executable, '-c', 'import time; time.sleep(30)'], timeout=0.1) self.assertIsNone(self.runner.child) def test_overlapping_backup_is_skipped(self): import fcntl with (self.runner.state/'lock').open('a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) with patch.object(self.runner, 'perform') as perform: self.runner.backup() perform.assert_not_called() def test_independent_hash_check_detects_same_size_corruption(self): import hashlib self.runner.stage.mkdir() (self.runner.stage/'file').write_bytes(b'wrong') manifest = {'file': {'size': 5, 'hashes': {'SHA-1': hashlib.sha1(b'right').hexdigest()}}} with self.assertRaisesRegex(m.BackupError, 'hash mismatch'): self.runner.verify_local_hashes(manifest) def test_disappearing_hash_requires_download_comparison(self): before = {'file': {'size': 10, 'mtime': 'now', 'hashes': {'SHA-1': 'abc'}}} after = {'file': {'size': 10, 'mtime': 'now', 'hashes': {}}} with patch.object(self.runner, 'verify_local_hashes') as local, \ patch.object(self.runner, 'cloud') as cloud: self.runner.validate_final_inventory(before, after) local.assert_called_once_with(after) self.assertIn('--download', cloud.call_args.args) def test_new_hash_with_wrong_contents_is_rejected(self): before = {'file': {'size': 10, 'mtime': 'now', 'hashes': {}}} after = {'file': {'size': 10, 'mtime': 'now', 'hashes': {'SHA-1': 'abc'}}} with patch.object(self.runner, 'verify_local_hashes', side_effect=m.BackupError('hash mismatch')): with self.assertRaises(m.BackupError): self.runner.validate_final_inventory(before, after) def test_installer_preserves_config_and_key_and_pauses_timer(self): home = self.root/'install-home' config = home/'.config/nextcloud-borg-backup' config.mkdir(parents=True) (config/'config').write_text('# preserved custom config\n') (config/'passphrase').write_text('existing-test-key') fake = self.root/'fake-bin' fake.mkdir() commands = self.root/'systemctl-calls' (fake/'systemctl').write_text('#!/bin/sh\nprintf "%s\\n" "$*" >> "$CALL_LOG"\nexit 0\n') (fake/'systemctl').chmod(0o755) env = dict(os.environ, HOME=str(home), XDG_CONFIG_HOME=str(home/'.config'), XDG_STATE_HOME=str(home/'.local/state'), PATH=str(fake)+':'+os.environ['PATH'], CALL_LOG=str(commands)) subprocess.run([str(ROOT/'scripts/install.sh')], env=env, check=True, stdout=subprocess.DEVNULL) self.assertEqual((config/'config').read_text(), '# preserved custom config\n') self.assertEqual((config/'passphrase').read_text(), 'existing-test-key') calls = commands.read_text() self.assertIn('stop nextcloud-borg-backup.service nextcloud-borg-backup.timer', calls) self.assertIn('restart nextcloud-borg-backup.timer nextcloud-borg-health.timer', calls) self.assertEqual((home/'.local/libexec/nextcloud-borg-runner.py').read_bytes(), (ROOT/'scripts/nextcloud-borg-runner.py').read_bytes()) self.assertEqual(list(home.rglob('*.new')), []) def test_pending_maintenance_survives_interrupted_attempt_record(self): import time self.runner.success = {'completed_at': time.time(), 'maintenance_pending': True} m.atomic_json(self.runner.attempt_file, {'status': 'running'}) with patch.object(self.runner, 'nas'), patch.object(self.runner, 'maintenance') as maintenance: self.runner.backup(True) maintenance.assert_called_once() def test_real_rclone_borg_backup_and_restore(self): source = self.root/'source' source.mkdir() (source/'hello.txt').write_text('recoverable cloud data\n') (source/'sub').mkdir() (source/'sub/unicode รค.txt').write_bytes(b'\x00\x01'*2048) subprocess.run(['borg', 'init', '--encryption=none', str(self.runner.repo)], check=True, stdout=subprocess.DEVNULL) # Only the host mount probe is mocked; real rclone, Borg create/check and restore run. with patch.object(self.runner, 'nas'), patch.object(self.runner, 'staging_preflight', side_effect=lambda _: self.runner.stage.mkdir()): self.runner.backup() success = m.read_json(self.runner.success_file) self.assertEqual(success['files'], 2) self.assertEqual(success['stage'], str(self.runner.stage)) dest = self.root/'restore' dest.mkdir() subprocess.run(['borg', 'extract', f'{self.runner.repo}::{success["archive"]}'], cwd=dest, check=True) for path in source.rglob('*'): if path.is_file(): self.assertEqual(path.read_bytes(), (dest/path.relative_to(source)).read_bytes()) # Same-size, same-time edits must work with hashes AND without server hashes. original_manifest = self.runner.manifest for day, no_hashes in [(1, False), (2, True)]: old = (source/'hello.txt').stat() (source/'hello.txt').write_text(f'{day}ECOVERABLE cloud data\n') os.utime(source/'hello.txt', ns=(old.st_atime_ns, old.st_mtime_ns)) def manifest(): result = original_manifest() if no_hashes: for value in result.values(): value['hashes'] = {} return result with patch.object(self.runner, 'nas'), patch.object(self.runner, 'staging_preflight'), \ patch.object(self.runner, 'manifest', side_effect=manifest), \ patch.object(m.dt, 'datetime', wraps=m.dt.datetime) as clock: clock.now.return_value = m.dt.datetime(2030, 1, day) self.runner.backup() self.assertEqual((source/'hello.txt').read_bytes(), (self.runner.stage/'hello.txt').read_bytes()) # A refused staging deletion must preserve the last verified success/archive. previous = m.read_json(self.runner.success_file) (source/'hello.txt').unlink() with patch.dict(os.environ, {'MAX_STAGE_DELETES': '0', 'MAX_SOURCE_DROP_PERCENT': '100'}), \ patch.object(self.runner, 'nas'), patch.object(self.runner, 'staging_preflight'), \ patch.object(self.runner, 'notify'): with self.assertRaises(m.BackupError): self.runner.backup() self.assertEqual(m.read_json(self.runner.success_file), previous) if __name__ == '__main__': unittest.main()