424 lines
21 KiB
Python
424 lines
21 KiB
Python
"""Integration tests use real tar/rsync/rclone crypt and a real isolated MariaDB.
|
|
Only systemd/occ are simulated: production services never receive test commands.
|
|
"""
|
|
import gzip
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location("backup_set", REPO / "lib/backup_set.py")
|
|
HELPER = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(HELPER)
|
|
|
|
|
|
class ArchiveValidation(unittest.TestCase):
|
|
def archive(self, entries):
|
|
temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temp.cleanup)
|
|
path = Path(temp.name) / 'test.tar.gz'
|
|
with tarfile.open(path, 'w:gz') as archive:
|
|
for name, kind, target in entries:
|
|
info = tarfile.TarInfo(name)
|
|
info.type = kind
|
|
info.linkname = target
|
|
archive.addfile(info)
|
|
return path
|
|
|
|
def test_traversal_and_link_pivot_rejected(self):
|
|
for entries in [
|
|
[('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/../../evil', tarfile.REGTYPE, '')],
|
|
[('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/link', tarfile.SYMTYPE, '/tmp'), ('files/wordpress/link/evil', tarfile.REGTYPE, '')],
|
|
[('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/hard', tarfile.LNKTYPE, '/etc/shadow')],
|
|
[('files/wordpress', tarfile.SYMTYPE, '/tmp')],
|
|
[('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/dev', tarfile.CHRTYPE, '')],
|
|
]:
|
|
with self.subTest(entries=entries), self.assertRaises(ValueError):
|
|
HELPER.check_tar(self.archive(entries), ['files/wordpress'])
|
|
|
|
def test_leaf_symlinks_allowed(self):
|
|
HELPER.check_tar(self.archive([('files/wordpress', tarfile.DIRTYPE, ''),
|
|
('files/wordpress/cert', tarfile.SYMTYPE, '/etc/ssl/cert.pem')]), ['files/wordpress'])
|
|
|
|
def test_truncated_compression_rejected(self):
|
|
path = self.archive([('files/wordpress', tarfile.DIRTYPE, '')])
|
|
path.write_bytes(path.read_bytes()[:-5])
|
|
with self.assertRaises((ValueError, tarfile.TarError)):
|
|
HELPER.check_tar(path, ['files/wordpress'])
|
|
|
|
def test_retention_keeps_minimum_and_ignores_v1(self):
|
|
result = subprocess.run(['python3', str(REPO/'lib/backup_set.py'), 'candidates', 'appbackup', '7', '2'],
|
|
input='appbackup_2020-01-01_00-00-00_aaaaaaaa\nappbackup_2020-01-02_00-00-00_bbbbbbbb\nappbackup_2020-01-03_00-00-00_cccccccc\nappbackup_2019-01-01_00-00-00\n../evil\n',
|
|
text=True, capture_output=True, check=True)
|
|
self.assertEqual(result.stdout.strip(), 'appbackup_2020-01-01_00-00-00_aaaaaaaa')
|
|
|
|
|
|
@unittest.skipUnless(os.geteuid() == 0 and shutil.which('mariadbd'), 'run integration tests in tests/Dockerfile container')
|
|
class Integration(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.dbtemp = tempfile.TemporaryDirectory(prefix='app-backup-mariadb-')
|
|
cls.dbdir = Path(cls.dbtemp.name)
|
|
cls.socket = cls.dbdir/'mysql.sock'
|
|
subprocess.run(['mariadb-install-db', '--no-defaults', f'--datadir={cls.dbdir}/data', '--auth-root-authentication-method=normal', '--skip-test-db'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
|
|
cls.server = subprocess.Popen(['mariadbd','--no-defaults','--user=root',f'--datadir={cls.dbdir}/data', f'--socket={cls.socket}', f'--pid-file={cls.dbdir}/pid', '--skip-networking'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
for _ in range(100):
|
|
if subprocess.run(['mariadb','--no-defaults',f'--socket={cls.socket}','-uroot','-e','SELECT 1'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0:
|
|
break
|
|
time.sleep(.1)
|
|
else:
|
|
raise RuntimeError('Test MariaDB did not start')
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.terminate()
|
|
cls.server.wait(timeout=20)
|
|
cls.dbtemp.cleanup()
|
|
|
|
def sql(self, query):
|
|
return subprocess.check_output(['mariadb','--no-defaults', f'--socket={self.socket}','-uroot','-N','-e',query], text=True)
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='app-backup-test-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.src = self.base/'sources'
|
|
self.wp = self.src/'web'
|
|
self.nc = self.wp/'cloud'
|
|
self.data = self.nc/'userdata'
|
|
self.bin = self.base/'bin'
|
|
self.bin.mkdir()
|
|
for p in [self.wp,self.nc,self.data,self.src/'gitea',self.src/'gitea-etc',self.src/'mail',self.src/'postfix',self.src/'dovecot']:
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
(p/'payload.txt').write_text(f'payload for {p.name}')
|
|
(self.wp/'nested/data').mkdir(parents=True)
|
|
(self.wp/'nested/data/keep.txt').write_text('nested data must survive')
|
|
(self.nc/'occ').write_text('test stub')
|
|
self.state=self.base/'state.json'
|
|
self.initial=dict(services={s:'active' for s in ['httpd','php-fpm','gitea','postfix','dovecot','nc-cron']}, maintenance=False)
|
|
self.state.write_text(json.dumps(self.initial))
|
|
self.events=self.base/'events'
|
|
self.events.touch()
|
|
mock='''#!/usr/bin/env python3
|
|
import json,os,sys
|
|
from pathlib import Path
|
|
name=Path(sys.argv[0]).name
|
|
args=sys.argv[1:]
|
|
p=Path(os.environ['TEST_STATE'])
|
|
s=json.loads(p.read_text())
|
|
with open(os.environ['TEST_EVENTS'],'a') as f: f.write(name+' '+' '.join(args)+'\\n')
|
|
if name=='systemctl':
|
|
cmd=args[0]
|
|
service=args[1] if len(args)>1 else ''
|
|
if cmd=='show': print(s['services'].get(service,'inactive'))
|
|
elif cmd=='is-active': sys.exit(0 if s['services'].get(args[-1])=='active' else 3)
|
|
elif cmd in ('stop','start'):
|
|
if os.environ.get('TEST_FAIL_SERVICE')==cmd+':'+service: sys.exit(1)
|
|
s['services'][service]='active' if cmd=='start' else 'inactive'
|
|
elif cmd in ('daemon-reload','enable'): pass
|
|
elif name=='sudo':
|
|
os.execvp(args[2],args[2:])
|
|
elif name=='php':
|
|
cmd=args[1]
|
|
if cmd=='status': print(json.dumps({'maintenance':s['maintenance']}))
|
|
elif cmd=='maintenance:mode': s['maintenance']='--on' in args
|
|
p.write_text(json.dumps(s))
|
|
'''
|
|
for name in ['systemctl','php','sudo']:
|
|
path=self.bin/name
|
|
path.write_text(mock)
|
|
path.chmod(0o755)
|
|
# Wrappers inject failures but normally delegate to the real tools.
|
|
for name in ['rsync','mariadb-dump','mariadb','zstd','rclone']:
|
|
real=shutil.which(name)
|
|
script=f'''#!/usr/bin/env bash
|
|
printf '%s\\n' '{name}' >> "$TEST_EVENTS"
|
|
if [[ ${{TEST_FAIL_TOOL:-}} == {name} ]]; then exit 42; fi
|
|
if [[ {name} == rclone && ${{TEST_FAIL_CHECK:-}} == true && " $* " == *' check '* ]]; then exit 43; fi
|
|
exec {real} "$@"
|
|
'''
|
|
path=self.bin/name
|
|
path.write_text(script)
|
|
path.chmod(0o755)
|
|
self.cnf=self.base/'db.cnf'
|
|
self.cnf.write_text(f'[client]\nuser=root\nsocket={self.socket}\n')
|
|
self.cnf.chmod(0o600)
|
|
self.sql('DROP DATABASE IF EXISTS backup_test; CREATE DATABASE backup_test; CREATE TABLE backup_test.t (id INT PRIMARY KEY, body TEXT) ENGINE=MyISAM; INSERT INTO backup_test.t VALUES (1,"original");')
|
|
self.config=self.base/'config'
|
|
self.config.write_text(f'''WORKDIR='{self.base}/work'
|
|
LOG_DIR='{self.base}/logs'
|
|
LOCKFILE='{self.base}/lock'
|
|
MIN_FREE_GB=0
|
|
COMPRESSOR=zstd
|
|
ENABLE_UPLOAD=false
|
|
ENABLE_MAIL_REPORT=false
|
|
ENABLE_REMOTE_RETENTION=false
|
|
ENABLE_WORDPRESS=true
|
|
WP_DIR='{self.wp}'
|
|
ENABLE_NEXTCLOUD=true
|
|
NC_DIR='{self.nc}'
|
|
NC_DATA_DIR='{self.data}'
|
|
NC_OCC_USER=root
|
|
ENABLE_GITEA=true
|
|
GITEA_DATA_DIR='{self.src}/gitea'
|
|
GITEA_ETC_DIR='{self.src}/gitea-etc'
|
|
ENABLE_MAIL=true
|
|
ENABLE_MAIL_HOME_DIRS=false
|
|
MAIL_DIR='{self.src}/mail'
|
|
POSTFIX_DIR='{self.src}/postfix'
|
|
DOVECOT_DIR='{self.src}/dovecot'
|
|
WP_DB_NAME=backup_test
|
|
NC_DB_NAME=backup_test
|
|
GITEA_DB_NAME=backup_test
|
|
WP_DB_CNF='{self.cnf}'
|
|
NC_DB_CNF='{self.cnf}'
|
|
GITEA_DB_CNF='{self.cnf}'
|
|
QUIESCE_SERVICES=(nc-cron)
|
|
''')
|
|
self.config.chmod(0o600)
|
|
self.env=dict(os.environ, PATH=str(self.bin)+':'+os.environ['PATH'],TEST_STATE=str(self.state),TEST_EVENTS=str(self.events))
|
|
|
|
def run_script(self, script, *args, ok=True, **env):
|
|
proc=subprocess.run(['bash',str(REPO/'usr-local-sbin'/script),'--config',str(self.config),*map(str,args)],env=dict(self.env,**env),text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=90)
|
|
if ok: self.assertEqual(proc.returncode,0,proc.stdout)
|
|
else: self.assertNotEqual(proc.returncode,0,proc.stdout)
|
|
return proc
|
|
|
|
def backup(self, **env):
|
|
self.run_script('app-backup.sh',**env)
|
|
return next((self.base/'work/archives').glob('appbackup_*'))
|
|
|
|
def test_real_database_and_files_round_trip(self):
|
|
run=self.backup()
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
data=HELPER.load_set(run)
|
|
self.assertEqual(len(data['archives']),12)
|
|
with subprocess.Popen(['zstd','-dc',str(run/'wordpress.tar.zst')],stdout=subprocess.PIPE) as p:
|
|
with tarfile.open(fileobj=p.stdout,mode='r|') as t: names=[m.name for m in t]
|
|
self.assertNotIn('files/wordpress/cloud',names)
|
|
self.assertIn('files/wordpress/nested/data/keep.txt',names)
|
|
self.sql('UPDATE backup_test.t SET body="changed"')
|
|
(self.wp/'payload.txt').write_text('changed')
|
|
self.events.write_text('')
|
|
self.run_script('app-restore.sh','--local-run',run,'--dry-run')
|
|
self.assertNotIn('systemctl stop',self.events.read_text())
|
|
self.assertEqual((self.wp/'payload.txt').read_text(),'changed')
|
|
self.run_script('app-restore.sh','--local-run',run,'--apply','--yes')
|
|
self.assertEqual(self.sql('SELECT body FROM backup_test.t').strip(),'original')
|
|
self.assertEqual((self.wp/'payload.txt').read_text(),'payload for web')
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
|
|
def test_dump_and_copy_failures_do_not_publish(self):
|
|
for tool in ['mariadb-dump','rsync','zstd']:
|
|
with self.subTest(tool=tool):
|
|
self.run_script('app-backup.sh',ok=False,TEST_FAIL_TOOL=tool)
|
|
self.assertFalse(list((self.base/'work/archives').glob('appbackup_*')))
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
|
|
def test_maintenance_and_inactive_service_preserved(self):
|
|
state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=True)
|
|
self.state.write_text(json.dumps(state))
|
|
self.backup()
|
|
self.assertEqual(json.loads(self.state.read_text()),state)
|
|
|
|
def test_corruption_and_missing_archive_block_restore(self):
|
|
run=self.backup()
|
|
path=run/'wordpress.tar.zst'
|
|
content=path.read_bytes()
|
|
path.write_bytes(content[:-10])
|
|
self.events.write_text('')
|
|
self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False)
|
|
self.assertNotIn('systemctl stop',self.events.read_text())
|
|
path.unlink()
|
|
self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False)
|
|
self.assertNotIn('systemctl stop',self.events.read_text())
|
|
|
|
def test_partial_restore_keeps_services_stopped(self):
|
|
run=self.backup()
|
|
self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False,TEST_FAIL_TOOL='mariadb')
|
|
# Connection preflight fails BEFORE mutation: service state stays unchanged.
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
wrapper=self.bin/'mariadb'
|
|
wrapper.write_text('#!/usr/bin/env bash\nif [[ " $* " != *" --execute=SELECT 1 "* ]]; then exit 44; fi\nexec /usr/bin/mariadb "$@"\n')
|
|
result=self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False)
|
|
self.assertIn('Partial restore',result.stdout)
|
|
self.assertTrue(json.loads(self.state.read_text())['maintenance'])
|
|
self.assertEqual(json.loads(self.state.read_text())['services']['httpd'],'inactive')
|
|
|
|
def test_check_does_not_claim_successful_backup(self):
|
|
self.run_script('app-backup.sh','--check')
|
|
self.assertFalse((self.base/'work/last-successful-backup.json').exists())
|
|
self.assertNotIn('systemctl stop',self.events.read_text())
|
|
|
|
def test_service_stop_and_start_failures_are_reported(self):
|
|
self.run_script('app-backup.sh',ok=False,TEST_FAIL_SERVICE='stop:gitea')
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
self.run_script('app-backup.sh',ok=False,TEST_FAIL_SERVICE='start:gitea')
|
|
self.assertTrue(list((self.base/'work').glob('recovery-*.json')))
|
|
self.assertNotEqual(json.loads((self.base/'work/last-backup.json').read_text())['exit_code'],0)
|
|
self.assertFalse(list((self.base/'work/archives').glob('appbackup_*')))
|
|
|
|
def test_term_recovers_services(self):
|
|
wrapper=self.bin/'rsync'
|
|
wrapper.write_text('#!/usr/bin/env bash\necho waiting > "$TEST_EVENTS.waiting"\nsleep 2\nexec /usr/bin/rsync "$@"\n')
|
|
proc=subprocess.Popen(['bash',str(REPO/'usr-local-sbin/app-backup.sh'),'--config',str(self.config)],env=self.env,text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
|
|
try:
|
|
for _ in range(100):
|
|
if Path(str(self.events)+'.waiting').exists(): break
|
|
time.sleep(.05)
|
|
else: self.fail('Backup did not reach rsync')
|
|
proc.terminate()
|
|
output=proc.communicate(timeout=20)[0]
|
|
self.assertEqual(proc.returncode,143,output)
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
self.assertFalse(list((self.base/'work/archives').glob('appbackup_*')))
|
|
finally:
|
|
if proc.poll() is None: proc.kill(); proc.wait()
|
|
|
|
def test_selected_strict_restore_preserves_excluded_nextcloud(self):
|
|
run=self.backup()
|
|
(self.wp/'extra.txt').write_text('delete me')
|
|
(self.data/'outside-selection.txt').write_text('keep me')
|
|
with self.config.open('a') as f: f.write('RESTORE_STRICT_DELETE=true\n')
|
|
self.run_script('app-restore.sh','--local-run',run,'--apply','--yes','--no-db','--components','wordpress')
|
|
self.assertFalse((self.wp/'extra.txt').exists())
|
|
self.assertEqual((self.data/'outside-selection.txt').read_text(),'keep me')
|
|
self.assertEqual(json.loads(self.state.read_text()),self.initial)
|
|
|
|
def test_local_and_remote_retention_preserve_partial_and_v1(self):
|
|
self.crypt_config()
|
|
current=self.backup()
|
|
archives=current.parent
|
|
old=[]
|
|
for day in range(1,4):
|
|
name=f'appbackup_2020-01-0{day}_00-00-00_aaaaaaaa'
|
|
path=archives/name
|
|
shutil.copytree(current,path)
|
|
manifest=json.loads((path/'manifest.json').read_text())
|
|
manifest['run_id']=name
|
|
(path/'manifest.json').write_text(json.dumps(manifest))
|
|
(path/'COMPLETE').write_text('APP_BACKUP_V2 '+HELPER.digest(path/'manifest.json')+'\n')
|
|
subprocess.run(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'),'copy',str(path),'crypt:sets/'+name],check=True,capture_output=True)
|
|
old.append(path)
|
|
partial=archives/'appbackup_2019-01-01_00-00-00_aaaaaaaa'
|
|
partial.mkdir()
|
|
v1=archives/'appbackup_2019-01-01_00-00-00'
|
|
v1.mkdir()
|
|
with self.config.open('a') as f: f.write('ENABLE_REMOTE_RETENTION=true\nRETENTION_KEEP_LAST=2\nLOCAL_RETENTION_DAYS=1\nREMOTE_RETENTION_DAYS=1\n')
|
|
self.run_script('app-backup.sh')
|
|
self.assertTrue(current.is_dir())
|
|
self.assertTrue(partial.is_dir())
|
|
self.assertTrue(v1.is_dir())
|
|
self.assertTrue(all(not p.exists() for p in old))
|
|
remote=subprocess.check_output(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'),'lsf','crypt:sets','--dirs-only'],text=True)
|
|
self.assertNotIn('2020-',remote)
|
|
self.assertIn(current.name,remote)
|
|
|
|
def test_update_preserves_configuration_schedule_and_rolls_back(self):
|
|
# Container-only test of the actual absolute installation paths.
|
|
config=Path('/etc/app-backup/app-backup.conf')
|
|
config.parent.mkdir(exist_ok=True)
|
|
original=f'LOCKFILE="{self.base}/install.lock"\n# existing custom configuration\n'
|
|
config.write_text(original)
|
|
config.chmod(0o600)
|
|
old_script=Path('/usr/local/sbin/app-backup.sh')
|
|
old_script.parent.mkdir(parents=True,exist_ok=True)
|
|
old_script.write_text('#!/bin/bash\necho previous-version\n')
|
|
old_script.chmod(0o750)
|
|
timer=Path('/etc/systemd/system/app-backup.timer')
|
|
timer.parent.mkdir(parents=True,exist_ok=True)
|
|
timer_text='[Timer]\nOnCalendar=*-*-* 04:10:00\n'
|
|
timer.write_text(timer_text)
|
|
state=json.loads(self.state.read_text())
|
|
state['services']['app-backup.timer']='active'
|
|
self.state.write_text(json.dumps(state))
|
|
proc=subprocess.run(['bash',str(REPO/'install.sh')],env=self.env,capture_output=True,text=True)
|
|
self.assertEqual(proc.returncode,0,proc.stdout+proc.stderr)
|
|
self.assertEqual(config.read_text(),original)
|
|
self.assertEqual(timer.read_text(),timer_text)
|
|
self.assertTrue(Path(str(config)+'.dist').exists())
|
|
self.assertEqual(old_script.read_text(),(REPO/'usr-local-sbin/app-backup.sh').read_text())
|
|
self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active')
|
|
snapshot=Path(proc.stdout.split('Rollback package: ')[1].splitlines()[0])
|
|
rollback=subprocess.run(['bash',str(snapshot/'rollback.sh')],env=self.env,capture_output=True,text=True)
|
|
self.assertEqual(rollback.returncode,0,rollback.stdout+rollback.stderr)
|
|
self.assertIn('previous-version',old_script.read_text())
|
|
self.assertEqual(config.read_text(),original)
|
|
self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'inactive')
|
|
|
|
def test_legacy_extraction_is_explicit_and_isolated(self):
|
|
folder=self.base/'legacy'
|
|
folder.mkdir()
|
|
archive=folder/'appbackup_2026-02-11_02-30-00_wordpress.tar.gz'
|
|
with tarfile.open(archive,'w:gz') as tar:
|
|
tar.add(self.wp,arcname='files/wordpress')
|
|
self.run_script('app-restore.sh','--local-run',folder,'--extract-only',ok=False)
|
|
self.run_script('app-restore.sh','--local-run',folder,'--legacy-unverified','--extract-only')
|
|
self.assertTrue(list((self.base/'work/restore').glob('*/extracted/*.tar.gz/files/wordpress/payload.txt')))
|
|
self.run_script('app-restore.sh','--local-run',folder,'--legacy-unverified','--apply','--yes',ok=False)
|
|
self.assertNotIn('systemctl stop',self.events.read_text())
|
|
|
|
def test_workspace_overlap_does_not_change_source_permissions(self):
|
|
self.wp.chmod(0o755)
|
|
with self.config.open('a') as f: f.write(f'WORKDIR="{self.wp}"\n')
|
|
self.run_script('app-backup.sh','--check',ok=False)
|
|
self.assertEqual(self.wp.stat().st_mode & 0o777,0o755)
|
|
|
|
def crypt_config(self):
|
|
(self.base/'encrypted').mkdir()
|
|
password=subprocess.check_output(['/usr/bin/rclone','obscure','integration-test-only'],text=True).strip()
|
|
path=self.base/'rclone.conf'
|
|
path.write_text(f'[crypt]\ntype = crypt\nremote = {self.base}/encrypted\npassword = {password}\n')
|
|
path.chmod(0o600)
|
|
with self.config.open('a') as f: f.write(f'ENABLE_UPLOAD=true\nRCLONE_CONFIG="{path}"\nRCLONE_REMOTE_BASE="crypt:sets"\nRCLONE_RETRIES=1\nRCLONE_LOW_LEVEL_RETRIES=1\n')
|
|
return path
|
|
|
|
def test_real_crypt_upload_readback_remote_restore(self):
|
|
self.crypt_config()
|
|
run=self.backup()
|
|
self.assertTrue((run/'UPLOADED').is_file())
|
|
self.assertFalse(list((self.base/'encrypted').rglob('wordpress.tar.zst')))
|
|
self.run_script('app-restore.sh','--remote-run',run.name,'--verify-only')
|
|
|
|
def test_failed_readback_does_not_publish_remote_marker(self):
|
|
self.crypt_config()
|
|
self.run_script('app-backup.sh',ok=False,TEST_FAIL_CHECK='true')
|
|
run=next((self.base/'work/archives').glob('appbackup_*'))
|
|
self.assertFalse((run/'UPLOADED').exists())
|
|
proc=subprocess.run(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'),'cat',f'crypt:sets/{run.name}/COMPLETE'],capture_output=True)
|
|
self.assertNotEqual(proc.returncode,0)
|
|
|
|
def test_unsafe_crypt_and_source_overlap_rejected(self):
|
|
path=self.crypt_config()
|
|
with path.open('a') as f: f.write('no_data_encryption = true\n')
|
|
self.run_script('app-backup.sh','--check',ok=False)
|
|
self.assertNotIn('systemctl stop',self.events.read_text())
|
|
with self.config.open('a') as f: f.write(f'ENABLE_UPLOAD=false\nWP_DIR="{self.base}"\n')
|
|
self.run_script('app-backup.sh','--check',ok=False)
|
|
|
|
def test_lock_excludes_concurrent_run(self):
|
|
import fcntl
|
|
with (self.base/'lock').open('w') as lock:
|
|
fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
result=self.run_script('app-backup.sh','--check',ok=False)
|
|
self.assertIn('Another backup/restore/update',result.stdout)
|
|
|
|
def test_gzip_and_extract_only(self):
|
|
with self.config.open('a') as f: f.write('COMPRESSOR=gzip\n')
|
|
run=self.backup()
|
|
self.run_script('app-restore.sh','--local-run',run,'--extract-only')
|
|
self.assertTrue(list((self.base/'work/restore').glob('*/extracted/wordpress/files/wordpress/payload.txt')))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|