cesnimda/jobtrackingapp
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(release): stop deployment when off-host backup transfer fails

Browse Source
This commit is contained in:
cesnimda
2026-11-10 13:24:30 +01:00
parent e73681008a
commit cedf33a358
5 changed files with 90 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/test-deploy-backup.py
+69 -1
View File
@@ -1,4 +1,4 @@
"""Exercise the real backup function with a synthetic dump command; never deploy."""
"""Exercise deployment backup functions with synthetic commands; never deploy."""
import os
from pathlib import Path
import shlex
@@ -85,5 +85,73 @@ class BackupConnectionTests(unittest.TestCase):
self.assertNotIn('dummy-secret', result.stdout + result.stderr)
class BackupUploadTests(unittest.TestCase):
def upload(self, fail_operation=''):
source = (ROOT / 'deploy/deploy.sh').read_text(encoding='utf-8')
start = source.index('upload_current_backup_set() {')
function = source[start:source.index('\n}\n', start) + 3]
gate_start = source.index('if ! apply_backup_retention || ! upload_current_backup_set; then')
gate = source[gate_start:source.index('\nfi\n', gate_start) + 4]
with tempfile.TemporaryDirectory() as temporary:
folder = Path(temporary)
for filename in ['jobtracker-db-synthetic.sql.gz', 'jobtracker-files-synthetic.tar.gz',
'jobtracker-db-older.sql.gz']:
(folder / filename).touch()
(folder / 'monthly' / '2099-01').mkdir(parents=True)
script = '\n'.join([
'set -euo pipefail',
'BACKUP_DIR=' + shlex.quote(folder.as_posix()),
'BACKUP_STAMP=synthetic',
'BACKUP_RCLONE_REMOTE=fixture:encrypted/',
'FAIL_OPERATION=' + shlex.quote(fail_operation),
'date() { echo 2099-01; }',
'apply_backup_retention() { return 0; }',
'rclone() {',
' printf "UPLOAD %s %s %s\\n" "$1" "$3" "$4";',
' if [ "$1" = "$FAIL_OPERATION" ]; then return 23; fi;',
'}',
function,
gate,
'echo "DEPLOY WOULD CONTINUE"',
])
result = subprocess.run([BASH, '--noprofile', '--norc'], input=script,
text=True, encoding='utf-8', capture_output=True,
cwd=ROOT, timeout=10,
env={key: value for key, value in os.environ.items()
if key.upper() in {'PATH', 'SYSTEMROOT', 'WINDIR', 'TEMP', 'TMP'}})
calls = [line for line in result.stdout.splitlines() if line.startswith('UPLOAD ')]
return result, calls
def test_uploads_only_current_daily_files_then_monthly_snapshot(self):
result, calls = self.upload()
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
self.assertEqual([
'UPLOAD copyto fixture:encrypted/daily/jobtracker-db-synthetic.sql.gz --immutable',
'UPLOAD copyto fixture:encrypted/daily/jobtracker-files-synthetic.tar.gz --immutable',
'UPLOAD copy fixture:encrypted/monthly/2099-01 --immutable',
], calls)
self.assertIn('DEPLOY WOULD CONTINUE', result.stdout)
self.assertIn('Current recovery set copied to configured off-host remote.', result.stdout)
def test_daily_copy_failure_stops_before_remaining_files_and_monthly_copy(self):
result, calls = self.upload('copyto')
self.assertEqual(1, result.returncode, result.stdout + result.stderr)
self.assertEqual([
'UPLOAD copyto fixture:encrypted/daily/jobtracker-db-synthetic.sql.gz --immutable',
], calls)
self.assertIn('Daily off-host backup upload failed', result.stdout)
self.assertNotIn('DEPLOY WOULD CONTINUE', result.stdout)
self.assertNotIn('Current recovery set copied', result.stdout)
def test_monthly_copy_failure_blocks_deploy(self):
result, calls = self.upload('copy')
self.assertEqual(1, result.returncode, result.stdout + result.stderr)
self.assertEqual(3, len(calls))
self.assertEqual('UPLOAD copy fixture:encrypted/monthly/2099-01 --immutable', calls[-1])
self.assertIn('Monthly off-host backup upload failed', result.stdout)
self.assertNotIn('DEPLOY WOULD CONTINUE', result.stdout)
self.assertNotIn('Current recovery set copied', result.stdout)
if __name__ == '__main__':
unittest.main()