53 lines
3.2 KiB
Python
53 lines
3.2 KiB
Python
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPT=Path(__file__).with_name('pre-backup.sh')
|
|
|
|
class BackupHookTests(unittest.TestCase):
|
|
def exercise(self, succeeds):
|
|
with tempfile.TemporaryDirectory(prefix='backup-hook-test-') as d:
|
|
root=Path(d); stage=root/'stage'; stage.mkdir(); binpath=root/'bin';binpath.mkdir()
|
|
previous=stage/'vaultwarden.pg_dump';previous.write_bytes(b'previous-good-backup')
|
|
docker=binpath/'docker';docker.write_text('#!/bin/sh\n[ "$1" = inspect ] && [ "$2" = vaultwarden ]\n');docker.chmod(0o755)
|
|
pg=binpath/'pg_dump';pg.write_text('#!/bin/sh\nprintf new-dump\nexit '+('0' if succeeds else '1')+'\n');pg.chmod(0o755)
|
|
env=dict(os.environ,PATH=str(binpath)+':'+os.environ['PATH'],RESTIC_STAGE_DIR=str(stage),RESTIC_DB_CREDS_FILE=str(root/'absent'),VW_PGPASS='fake',VW_PGHOST='fake',VW_PGUSER='fake',VW_PGDB='fake')
|
|
r=subprocess.run(['bash',str(SCRIPT)],env=env,capture_output=True,text=True)
|
|
if succeeds:
|
|
self.assertEqual(r.returncode,0,r.stdout+r.stderr)
|
|
self.assertEqual(previous.read_bytes(),b'new-dump')
|
|
else:
|
|
self.assertNotEqual(r.returncode,0,r.stdout+r.stderr)
|
|
self.assertEqual(previous.read_bytes(),b'previous-good-backup')
|
|
self.assertIn('required database dump(s) failed',r.stdout)
|
|
self.assertEqual(list(stage.glob('.pending.*')),[])
|
|
|
|
def test_failed_required_dump_preserves_previous_backup_and_cleans_scratch(self):
|
|
self.exercise(False)
|
|
|
|
def test_success_publishes_new_dump_and_cleans_scratch(self):
|
|
self.exercise(True)
|
|
|
|
def test_gitea_failure_removes_sql_scratch_and_exposes_error(self):
|
|
with tempfile.TemporaryDirectory(prefix='backup-gitea-test-') as d:
|
|
root=Path(d);stage=root/'stage';stage.mkdir();binpath=root/'bin';binpath.mkdir()
|
|
previous=stage/'gitea-dump.tar';previous.write_bytes(b'previous-good-archive')
|
|
docker=binpath/'docker'
|
|
docker.write_text('#!/bin/sh\nif [ "$1" = inspect ]; then [ "$2" = gitea ]; exit $?; fi\nshift 4\nexec "$@"\n')
|
|
docker.chmod(0o755)
|
|
gitea=binpath/'gitea'
|
|
gitea.write_text('#!/bin/sh\nwhile [ "$#" -gt 0 ]; do if [ "$1" = --tempdir ]; then shift; scratch=$1; fi; shift; done\nprintf %s "$scratch" > "$TEST_SCRATCH_PATH"\nprintf partial-sql > "$scratch/gitea-db.sql123"\necho simulated-export-failure >&2\nexit 9\n')
|
|
gitea.chmod(0o755)
|
|
path_record=root/'scratch-path'
|
|
env=dict(os.environ,PATH=str(binpath)+':'+os.environ['PATH'],RESTIC_STAGE_DIR=str(stage),RESTIC_DB_CREDS_FILE=str(root/'absent'),TEST_SCRATCH_PATH=str(path_record))
|
|
r=subprocess.run(['bash',str(SCRIPT)],env=env,capture_output=True,text=True)
|
|
self.assertNotEqual(r.returncode,0)
|
|
self.assertIn('simulated-export-failure',r.stderr)
|
|
self.assertFalse(Path(path_record.read_text()).exists())
|
|
self.assertEqual(previous.read_bytes(),b'previous-good-archive')
|
|
self.assertEqual(list(stage.glob('.pending.*')),[])
|
|
|
|
if __name__=='__main__':unittest.main()
|