"""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_missing_mail_store_requires_explicit_resolution(self): canonical=self.configure_upgrade() with canonical.open('a') as f: f.write(f'MAIL_DIR="{self.src}/missing-mail"\n') original=canonical.read_bytes() 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) self.assertIn('--mail-dir DIR',result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertFalse((self.src/'missing-mail').exists()) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_package_mail_directory_override(self): canonical=self.configure_upgrade() with canonical.open('a') as f: f.write(f'MAIL_DIR="{self.src}/missing-mail"\n') package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground','--mail-dir',str(self.src/'mail')],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('ENABLE_MAIL_STORE=true',canonical.read_text()) run=next((self.base/'work/archives').glob('appbackup_*')) self.assertTrue((run/'mail.tar.zst').is_file()) self.assertTrue((run/'UPLOADED').is_file()) def test_package_homes_only_keeps_mail_configs_and_service_stops(self): canonical=self.configure_upgrade() home=self.src/'johannes' for folder in ['cur','new','tmp']: (home/'Mail'/folder).mkdir(parents=True) mailbox=home/'Mail/cur/test-message:2,S' mailbox.write_bytes(b'Subject: retained mail\n\nMailbox content\n') mailbox.chmod(0o600) original=mailbox.read_bytes() with canonical.open('a') as f: f.write(f'MAIL_DIR="{self.src}/missing-mail"\nENABLE_MAIL_HOME_DIRS=true\nMAIL_HOME_DIRS=("{home}")\n') package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground','--mail-homes-only'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('ENABLE_MAIL_STORE=false',canonical.read_text()) run=next((self.base/'work/archives').glob('appbackup_*')) for name in ['mail-home-0','postfix','dovecot']: self.assertTrue((run/f'{name}.tar.zst').is_file()) self.assertFalse((run/'mail.tar.zst').exists()) self.assertTrue((run/'UPLOADED').is_file()) for unit in ['postfix','dovecot']: self.assertIn(f'systemctl stop {unit}',self.events.read_text()) self.assertEqual(json.loads(self.state.read_text())['services'][unit],'active') mailbox.write_text('changed after backup') restored=subprocess.run(['bash',str(REPO/'usr-local-sbin/app-restore.sh'),'--config',str(canonical), '--local-run',str(run),'--components','mail-home-0','--apply','--yes'], env=self.env,capture_output=True,text=True,timeout=60) self.assertEqual(restored.returncode,0,restored.stdout+restored.stderr) self.assertEqual(mailbox.read_bytes(),original) self.assertEqual(mailbox.stat().st_mode & 0o777,0o600) def test_homes_only_rejects_disabled_empty_and_missing_homes(self): canonical=self.configure_upgrade() original=canonical.read_text() package=self.build_package() for settings in ['ENABLE_MAIL_HOME_DIRS=false', 'ENABLE_MAIL_HOME_DIRS=true\nMAIL_HOME_DIRS=()', f'ENABLE_MAIL_HOME_DIRS=true\nMAIL_HOME_DIRS=("{self.src}/missing-home")']: with self.subTest(settings=settings): canonical.write_text(original+'\n'+settings+'\n') before=canonical.read_bytes() result=subprocess.run(['bash',str(package),'--foreground','--mail-homes-only'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),before) 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 systemd_launcher(self): """Model start-job waiting, detached FDs, and ambiguous submit failures.""" 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 assert '--property=Type=oneshot' in args command=args[args.index('/bin/bash'):] state=Path(command[-1]) if os.environ.get('TEST_SYSTEMD_SUBMIT') == 'reject': sys.exit('systemd refused submission') with open(state/'test-worker.log','w') as log: worker=subprocess.Popen(command,stdin=subprocess.DEVNULL,stdout=log,stderr=subprocess.STDOUT,env=os.environ,close_fds=True,start_new_session=True) if os.environ.get('TEST_SYSTEMD_SUBMIT') == 'accepted-error': sys.exit('simulated reply failure after systemd accepted the job') # Type=oneshot remains starting until ExecStart exits. Real systemd-run waits # for that start job unless --no-block is used; our old stub missed this. if '--no-block' not in args: try: sys.exit(worker.wait(timeout=3)) except subprocess.TimeoutExpired: worker.terminate() worker.wait(timeout=5) sys.exit('oneshot start job still waiting: launcher/worker lock deadlock') ''') launcher.chmod(0o755) def test_detached_package_handoff_finishes_after_starter_exits(self): self.configure_upgrade() self.systemd_launcher() 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_systemd_submission_failure_preserves_installation_and_reports_result(self): canonical=self.configure_upgrade() original=canonical.read_bytes() self.systemd_launcher() package=self.build_package() result=subprocess.run(['bash',str(package)],env=dict(self.env,TEST_SYSTEMD_SUBMIT='reject'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertIn('FAILED: systemd submission failed',(state/'result.txt').read_text()) self.assertIn(str(state/'result.txt'),result.stdout) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_failed_submission_revokes_even_an_already_queued_worker(self): canonical=self.configure_upgrade() original=canonical.read_bytes() self.systemd_launcher() package=self.build_package() result=subprocess.run(['bash',str(package)],env=dict(self.env,TEST_SYSTEMD_SUBMIT='accepted-error'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) for _ in range(100): log=(state/'test-worker.log').read_text() if 'Installation not started' in log: break time.sleep(.1) self.assertIn('submission was cancelled',log) self.assertIn('Installation not started',log) self.assertTrue((state/'result.txt').read_text().startswith('FAILED')) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) self.assertFalse((state/'package-rollback').exists()) def test_worker_lock_timeout_is_persisted_before_installation(self): import fcntl canonical=self.configure_upgrade() original=canonical.read_bytes() self.systemd_launcher() package=self.build_package() # Obtain a complete prepared state without starting an actual worker. result=subprocess.run(['bash',str(package)],env=dict(self.env,TEST_SYSTEMD_SUBMIT='reject'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) (state/'result.txt').write_text('QUEUED\n') wrapper=self.bin/'flock' # Keep real flock contention; shorten only the 60s timeout in this test. wrapper.write_text('#!/usr/bin/env bash\nif [[ $1 == -w && $2 == 60 ]]; then shift 2; exec /usr/bin/flock -w 0 "$@"; fi\nexec /usr/bin/flock "$@"\n') wrapper.chmod(0o755) with open('/run/app-backup-upgrade.lock','a') as lock: fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB) failed=subprocess.run(['bash',str(REPO/'update.sh'),'--worker',str(state)],env=self.env,capture_output=True,text=True,timeout=10) self.assertNotEqual(failed.returncode,0) self.assertIn('Timed out waiting for update coordinator lock',(state/'update.log').read_text()) self.assertIn('Installation not started',(state/'result.txt').read_text()) self.assertTrue((state/'result.txt').read_text().startswith('FAILED')) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) 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()