239 lines
12 KiB
Python
239 lines
12 KiB
Python
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_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_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)
|
|
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()
|