cesnimda
83 lines
4.0 KiB
Python
83 lines
4.0 KiB
Python
"""Exercise the deployment exit handler with synthetic image references and no Docker."""
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BASH = 'C:/Program Files/Git/bin/bash.exe' if os.name == 'nt' else shutil.which('bash')
|
|
|
|
|
|
class DeployRollbackTests(unittest.TestCase):
|
|
def exit_deploy(self, status, *, replaced=True, backend=True, frontend=True, compose_status=0):
|
|
source = (ROOT / 'deploy/deploy.sh').read_text(encoding='utf-8')
|
|
start = source.index('finish_deploy() {')
|
|
handler = source[start:source.index('\ntrap finish_deploy EXIT\n', start) + len('\ntrap finish_deploy EXIT\n')]
|
|
script = '\n'.join([
|
|
'set -euo pipefail',
|
|
f'replacement_started={str(replaced).lower()}',
|
|
'previous_backend_ref=' + ('synthetic/backend:previous' if backend else ''),
|
|
'previous_frontend_ref=' + ('synthetic/frontend:previous' if frontend else ''),
|
|
'auth_config_body_file=synthetic-body',
|
|
'auth_config_headers_file=synthetic-headers',
|
|
'rm() { printf "CALL rm %s\\n" "$*"; }',
|
|
'docker() { echo "CALL unexpected docker"; return 99; }',
|
|
'curl() { echo "CALL unexpected curl"; return 99; }',
|
|
'compose() {',
|
|
' printf "CALL compose %s [%s] [%s]\\n" "$*" "${BACKEND_IMAGE:-}" "${FRONTEND_IMAGE:-}";',
|
|
f' if [ "$1" = up ]; then return {compose_status}; fi;',
|
|
' return 0;',
|
|
'}',
|
|
handler,
|
|
f'exit {status}',
|
|
])
|
|
# Exclude inherited deployment settings from the fixture.
|
|
environment = {key: value for key, value in os.environ.items()
|
|
if key.upper() in {'PATH', 'SYSTEMROOT', 'WINDIR', 'TEMP', 'TMP'}}
|
|
result = subprocess.run([BASH, '--noprofile', '--norc'], input=script,
|
|
text=True, encoding='utf-8', capture_output=True,
|
|
cwd=ROOT, env=environment, timeout=10)
|
|
self.assertEqual(status, result.returncode, result.stdout + result.stderr)
|
|
self.assertEqual('', result.stderr)
|
|
calls = [line for line in result.stdout.splitlines() if line.startswith('CALL ')]
|
|
self.assertEqual(['CALL rm -f synthetic-body', 'CALL rm -f synthetic-headers'], calls[:2])
|
|
return result.stdout, calls[2:]
|
|
|
|
def test_failed_replacement_recreates_only_prior_core_images(self):
|
|
stdout, calls = self.exit_deploy(41)
|
|
self.assertEqual([
|
|
'CALL compose up -d --force-recreate --remove-orphans backend frontend '
|
|
'[synthetic/backend:previous] [synthetic/frontend:previous]',
|
|
], calls)
|
|
self.assertIn('Deployment remains failed', stdout)
|
|
|
|
def test_rollback_failure_retains_original_status_and_collects_diagnostics(self):
|
|
stdout, calls = self.exit_deploy(41, compose_status=19)
|
|
self.assertEqual([
|
|
'CALL compose up -d --force-recreate --remove-orphans backend frontend '
|
|
'[synthetic/backend:previous] [synthetic/frontend:previous]',
|
|
'CALL compose ps [] []',
|
|
'CALL compose logs --tail=200 backend frontend [] []',
|
|
], calls)
|
|
self.assertIn('Manual recovery is required', stdout)
|
|
|
|
def test_incomplete_previous_release_does_not_recreate_any_service(self):
|
|
for backend, frontend in [(False, True), (True, False), (False, False)]:
|
|
with self.subTest(backend=backend, frontend=frontend):
|
|
stdout, calls = self.exit_deploy(41, backend=backend, frontend=frontend)
|
|
self.assertEqual([], calls)
|
|
self.assertIn('no complete previous core release', stdout)
|
|
|
|
def test_failure_before_replacement_does_not_recreate_any_service(self):
|
|
_, calls = self.exit_deploy(41, replaced=False)
|
|
self.assertEqual([], calls)
|
|
|
|
def test_success_does_not_recreate_any_service(self):
|
|
_, calls = self.exit_deploy(0)
|
|
self.assertEqual([], calls)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|