test(release): exercise rollback handler offline

2026-11-10 13:21:56 +01:00
parent 47c6834cee
commit e73681008a
4 changed files with 106 additions and 4 deletions
@@ -0,0 +1,82 @@
+"""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()