"""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_writer_drain_waits_and_reports_timeout(self): with tempfile.TemporaryDirectory() as directory: writer=subprocess.Popen(['python3','-c','import time; time.sleep(1)',directory+'/cron.php']) try: started=time.monotonic() HELPER.wait_nextcloud(directory,3) self.assertGreaterEqual(time.monotonic()-started,.5) finally: writer.wait(timeout=5) writer=subprocess.Popen(['python3','-c','import time; time.sleep(5)',directory+'/occ']) try: with self.assertRaisesRegex(ValueError,'writers still active'): HELPER.wait_nextcloud(directory,1) finally: writer.terminate() writer.wait(timeout=5) 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': if '--property=LoadState' in args: print('loaded' if service in s['services'] else 'not-found') elif '--property=Environment' in args or '--property=TriggeredBy' in args: print('') else: 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=='enable': s['services'][args[-1]]='active' elif cmd=='daemon-reload': 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 build_package(self): output=self.base/'package' subprocess.run(['python3',str(REPO/'tools/build-update-package.py'),'--version','integration','--output',str(output)],check=True,capture_output=True) return output/'app-backup-update-integration.run' def configure_upgrade(self): self.crypt_config() canonical=Path('/etc/app-backup/app-backup.conf') canonical.parent.mkdir(exist_ok=True) shutil.copyfile(self.config,canonical) canonical.chmod(0o600) # Old installations commonly left credential files group-readable. self.cnf.chmod(0o640) state=json.loads(self.state.read_text()) state['services']['app-backup.timer']='active' self.state.write_text(json.dumps(state)) return canonical def test_automatic_package_upgrade_and_combined_rollback(self): canonical=self.configure_upgrade() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertIn('SUCCESS',(state/'result.txt').read_text()) self.assertTrue((state/'verified-backup-run').is_file()) self.assertIn('ENABLE_REMOTE_RETENTION=false',canonical.read_text()) self.assertEqual(self.cnf.stat().st_mode & 0o777,0o600) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') rollback=subprocess.run(['bash',str(state/'rollback-update.sh')],env=self.env,capture_output=True,text=True,timeout=30) self.assertEqual(rollback.returncode,0,rollback.stdout+rollback.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'inactive') def test_package_preflight_failure_preserves_live_configuration_and_timer(self): canonical=self.configure_upgrade() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=dict(self.env,TEST_FAIL_TOOL='mariadb'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(self.cnf.stat().st_mode & 0o777,0o640) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_package_failed_backup_keeps_timer_stopped_and_rollback_available(self): canonical=self.configure_upgrade() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=dict(self.env,TEST_FAIL_TOOL='mariadb-dump'),capture_output=True,text=True,timeout=120) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertIn('FAILED',(state/'result.txt').read_text()) self.assertTrue((state/'rollback-update.sh').is_file()) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'inactive') self.assertFalse((state/'verified-backup-run').exists()) rollback=subprocess.run(['bash',str(state/'rollback-update.sh')],env=self.env,capture_output=True,text=True,timeout=30) self.assertEqual(rollback.returncode,0,rollback.stdout+rollback.stderr) self.assertEqual(canonical.read_bytes(),original) def test_package_corruption_is_rejected_before_preparation(self): package=self.build_package() original=package.read_bytes() package.write_bytes(original[:-1]+bytes([original[-1]^1])) result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=20) self.assertNotEqual(result.returncode,0) self.assertIn('Paket-Pruefsumme falsch',result.stderr) self.assertNotIn('systemctl',self.events.read_text()) def test_inherited_lock_flag_cannot_bypass_another_process(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,APP_BACKUP_INHERIT_LOCK='1') self.assertIn('Inherited lock descriptor',result.stdout) def test_detached_package_handoff_finishes_after_starter_exits(self): self.configure_upgrade() launcher=self.bin/'systemd-run' launcher.write_text('''#!/usr/bin/env python3 import os,subprocess,sys from pathlib import Path args=sys.argv[1:] assert '--unit=app-backup-upgrade' in args command=args[args.index('/bin/bash'):] state=Path(command[-1]) with open(state/'test-worker.log','w') as log: subprocess.Popen(command,stdin=subprocess.DEVNULL,stdout=log,stderr=subprocess.STDOUT,env=os.environ,close_fds=True,start_new_session=True) ''') launcher.chmod(0o755) package=self.build_package() result=subprocess.run(['bash',str(package)],env=self.env,capture_output=True,text=True,timeout=60) self.assertEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) for _ in range(120): status=(state/'result.txt').read_text() if (state/'result.txt').exists() else '' if status.startswith(('SUCCESS','FAILED')): break time.sleep(.5) self.assertTrue(status.startswith('SUCCESS'),status+(state/'test-worker.log').read_text()) self.assertIn('unabhaengig von SSH',result.stdout) def test_cron_inventory_requires_explicit_policy_and_sorts_timer_first(self): canonical=self.configure_upgrade() cron=Path('/etc/cron.d/app-backup-test-nextcloud') cron.parent.mkdir(exist_ok=True) cron.write_text(f'*/5 * * * * root /usr/bin/php -f {self.nc}/cron.php\n') self.addCleanup(cron.unlink) state=json.loads(self.state.read_text()) state['services']['crond.service']='active' state['services']['nc-cron.timer']='active' self.state.write_text(json.dumps(state)) package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('noninteractive',result.stdout+result.stderr) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') result=subprocess.run(['bash',str(package),'--foreground','--cron-services','nc-cron,crond.service,nc-cron.timer'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) text=canonical.read_text() line=[line for line in text.splitlines() if line.startswith('QUIESCE_SERVICES=')][-1] self.assertLess(line.index('nc-cron.timer'),line.index('crond.service')) self.assertEqual(json.loads(self.state.read_text())['services']['crond.service'],'active') 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()