"""Exercise the real deployment shell in an isolated Linux filesystem fixture. Run: python3 -m unittest discover -s ops/deploy/tests -p test_deploy_lifecycle.py Only absolute infrastructure paths and the root check are substituted. All deployment decisions, traps, cleanup, state writes and switching are real. """ import json import os from pathlib import Path import subprocess import tempfile import time import unittest SOURCE = Path(__file__).resolve().parents[1] / "eusrr-deploy" OLD, PREVIOUS, TARGET, ORPHAN = (c * 40 for c in "abcd") FAKE = r'''#!/usr/bin/env python3 import json, os, pathlib, shutil, sys, time p = pathlib.Path root = p(os.environ['FIXTURE']) name = p(sys.argv[0]).name args = sys.argv[1:] with (root/'commands').open('a') as f: f.write(name+' '+repr(args)+'\n') def fail(stage): if os.environ.get('FAIL_STAGE') == stage: sys.exit(17) if name == 'runuser': args = args[args.index('--')+1:] os.execvp(args[0], args) elif name == 'git': if 'rev-parse' in args: print(os.environ['TARGET']) elif 'remove' in args: shutil.rmtree(args[-1]) elif 'add' in args: release=p(args[-2]); (release/'backend').mkdir(parents=True) (release/'frontend/.next/static').mkdir(parents=True) (release/'frontend/.next/static/new.js').touch() sys.exit(0) elif name == 'df': print('Available') available = os.environ.get('FREE_BYTES','20000000000') if os.environ.get('FAIL_STAGE') == 'disk_during_build' and (root/'building').exists(): available = '100000000' print('1000000' if '--output=iavail' in args else available) elif name == 'systemctl': file=root/'services.json'; states=json.loads(file.read_text()) def units(): return [a for a in args[1:] if a.endswith('.service')] def save(): file.write_text(json.dumps(states)) if args[0] == 'show': prop=args[args.index('-p')+1] print('loaded' if prop=='LoadState' else states.get(args[1],'inactive')) elif args[0] == 'is-active': sys.exit(0 if all(states.get(u)=='active' for u in units()) else 3) elif args[0] in ('start','restart'): for u in units(): states[u]='active' save() if any('celery-beat@blue' in u for u in units()): fail('new_beat') elif args[0] == 'stop' or (args[0]=='disable' and '--now' in args): if args[0]=='stop' and os.environ.get('FAIL_STAGE')=='stop_blue': sys.exit(1) for u in units(): states[u]='inactive' save() elif args[0] == 'reload': fail('reload') elif args[0] == 'enable' and any('@blue' in u for u in units()): fail('enable') elif name == 'fake-python': if 'venv' in args: dest=p(args[-1])/'bin'; dest.mkdir(parents=True) (dest/'python').symlink_to(root/'bin/fake-python') elif 'pip' in args: fail('pip') elif name == 'python': if 'pip' in args: fail('pip') elif '-' in args: script=sys.stdin.read() if 'os.execve' not in script: sys.exit(19) p(os.environ['EUSRR_DB_DUMP']).write_bytes(b'dump') fail('dump') elif 'migrate' in args and '--noinput' in args: fail('migration') elif name == 'npm': if 'ci' in args: base=p(args[args.index('--prefix')+1])/'node_modules' (base/'lightningcss-linux-x64-gnu').mkdir(parents=True) (base/'lightningcss-linux-x64-gnu/lightningcss.linux-x64-gnu.node').touch() (base/'@next/swc-linux-x64-gnu').mkdir(parents=True) if 'build' in args: fail('build') if os.environ.get('FAIL_STAGE') in ('hang','disk_during_build'): (root/'building').touch(); time.sleep(60) elif name == 'curl': if os.environ.get('FAIL_STAGE')=='health' and any(':3001' in a for a in args): print('502') else: print('200') elif name == 'sleep': time.sleep(0.02) elif name == 'findmnt': print('/') ''' @unittest.skipUnless(os.name == "posix", "Linux shell integration tests") class DeploymentLifecycleTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="eusrr-deploy-test-") self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.bin = self.root / "bin" self.bin.mkdir() fake = self.bin / "fake" fake.write_text(FAKE) fake.chmod(0o755) for name in ('runuser', 'git', 'df', 'systemctl', 'fake-python', 'npm', 'curl', 'sleep', 'findmnt', 'nginx', 'rsync', 'pg_restore'): (self.bin / name).symlink_to(fake) text = SOURCE.read_text() for prefix in ('/srv/', '/var/', '/run/', '/etc/'): text = text.replace(prefix, str(self.root) + prefix) text = text.replace('if [[ ${EUID} -ne 0 ]]; then', 'if false; then') text = text.replace('/usr/bin/python3 -m venv', 'fake-python -m venv') # Fixture Python entrypoint must preserve argv[0] for pip/dump handlers. text = text.replace('install -d -o root -g root', 'install -d') text = text.replace('install -d -o "${DEPLOY_USER}" -g "${DEPLOY_USER}"', 'install -d') text = text.replace('install -o root -g root', 'install') self.script = self.root / "deploy" self.script.write_text(text) self.state = self.root / "var/lib/eusrr-deploy" self.releases = self.root / "srv/eusrr-releases" self.slots = self.root / "srv/eusrr-slots" for directory in (self.state, self.releases, self.slots, self.root/'var/log', self.root/'run/lock', self.root/'srv/eusrr/backend', self.root/'srv/eusrr-shared/next-static', self.root/'etc/nginx/conf.d', self.root/'var/backups/eusrr/deploy'): directory.mkdir(parents=True, exist_ok=True) for sha in (OLD, PREVIOUS, ORPHAN): (self.releases/sha/'frontend/.next/static').mkdir(parents=True) self.slots.joinpath('green').symlink_to(self.releases/OLD) self.slots.joinpath('blue').symlink_to(self.releases/PREVIOUS) (self.root/'srv/eusrr-current').symlink_to(self.releases/OLD) (self.state/'active-slot').write_text('green\n') (self.state/'last-successful-sha').write_text(OLD+'\n') (self.root/'etc/nginx/conf.d/eusrr-active-upstreams.conf').write_text( 'server 127.0.0.1:9002;\nserver 127.0.0.1:3002;\n') self.services = self.root / 'services.json' self.services.write_text(json.dumps({ **{f'eusrr-{unit}@green.service': 'active' for unit in ('backend', 'frontend', 'celery-worker', 'celery-beat')}, 'celery-eusrr.service': 'active', 'celery-eusrr-beat.service': 'active', 'celery-eusrr-worker.service': 'active', })) self.env = {**os.environ, 'FIXTURE': str(self.root), 'TARGET': TARGET, 'PATH': str(self.bin) + ':' + os.environ['PATH']} def run_deploy(self, mode=TARGET, **env): result = subprocess.run(['bash', str(self.script), mode], env={**self.env, **env}, capture_output=True, text=True, timeout=30) self.output = result.stdout + result.stderr return result def assert_old_live(self): self.assertEqual((self.root/'srv/eusrr-current').resolve(), self.releases/OLD, self.output) self.assertEqual((self.state/'active-slot').read_text().strip(), 'green', self.output) states = json.loads(self.services.read_text()) for unit in ('backend', 'frontend', 'celery-worker', 'celery-beat'): self.assertEqual(states[f'eusrr-{unit}@green.service'], 'active', self.output) def test_success_preserves_old_and_removes_older_releases(self): result = self.run_deploy() self.assertEqual(result.returncode, 0, self.output) self.assertEqual((self.root/'srv/eusrr-current').resolve(), self.releases/TARGET) self.assertTrue((self.releases/OLD).exists()) self.assertFalse((self.releases/PREVIOUS).exists()) self.assertFalse((self.releases/ORPHAN).exists()) self.assertFalse((self.state/'pending-deployment').exists()) states = json.loads(self.services.read_text()) self.assertEqual(states['celery-eusrr-beat.service'], 'inactive') self.assertEqual(states['eusrr-celery-beat@green.service'], 'inactive') self.assertEqual(states['eusrr-celery-beat@blue.service'], 'active') def test_failures_remove_candidate_preserve_both_successful_releases(self): for stage in ('pip', 'build', 'dump', 'migration', 'health', 'new_beat', 'enable'): with self.subTest(stage=stage): result = self.run_deploy(FAIL_STAGE=stage) self.assertNotEqual(result.returncode, 0, self.output) self.assert_old_live() self.assertFalse((self.releases/TARGET).exists(), self.output) self.assertTrue((self.releases/PREVIOUS).exists(), self.output) self.assertFalse((self.state/'pending-deployment').exists(), self.output) self.assertFalse(list((self.root/'var/backups/eusrr/deploy').glob('*.partial'))) def test_capacity_failure_does_not_remove_previous(self): result = self.run_deploy(FREE_BYTES='4000000000') self.assertEqual(result.returncode, 6, self.output) self.assert_old_live() self.assertTrue((self.releases/PREVIOUS).exists()) def test_watchdog_stops_build_before_disk_exhaustion(self): result = self.run_deploy(FAIL_STAGE='disk_during_build') self.assertNotEqual(result.returncode, 0, self.output) self.assertIn('Stopping build before', self.output) self.assert_old_live() self.assertFalse((self.releases/TARGET).exists(), self.output) def test_failed_nginx_recovery_retains_candidate_and_journal(self): self.assertNotEqual(self.run_deploy(FAIL_STAGE='reload').returncode, 0) self.assertTrue((self.releases/TARGET).exists(), self.output) self.assertTrue((self.state/'pending-deployment').exists(), self.output) self.assertTrue((self.state/f'upstreams-before-{TARGET}').exists(), self.output) states = json.loads(self.services.read_text()) self.assertEqual(states['eusrr-backend@blue.service'], 'active') self.assertEqual(states['eusrr-frontend@blue.service'], 'active') self.assertEqual(self.run_deploy('--maintenance').returncode, 4, self.output) self.assertTrue((self.releases/TARGET).exists()) def test_already_active_sha_never_removed(self): self.env['TARGET'] = OLD self.assertEqual(self.run_deploy(OLD).returncode, 0, self.output) self.assert_old_live() self.assertTrue((self.releases/OLD).exists()) def test_deploying_standby_sha_does_not_delete_rollback(self): self.env['TARGET'] = PREVIOUS self.assertNotEqual(self.run_deploy(PREVIOUS).returncode, 0) self.assert_old_live() self.assertTrue((self.releases/PREVIOUS).exists()) def test_existing_lock_prevents_any_maintenance(self): import fcntl with (self.root/'run/lock/eusrr-deploy.lock').open('w') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) self.assertEqual(self.run_deploy('--maintenance').returncode, 3, self.output) self.assertTrue((self.releases/ORPHAN).exists()) def test_inconsistent_state_refuses_cleanup(self): (self.state/'last-successful-sha').write_text(TARGET) self.assertNotEqual(self.run_deploy('--maintenance').returncode, 0) self.assertTrue((self.releases/ORPHAN).exists()) def test_failure_to_stop_refuses_cleanup(self): self.assertNotEqual(self.run_deploy('--maintenance', FAIL_STAGE='stop_blue').returncode, 0) self.assertTrue((self.releases/ORPHAN).exists()) def test_maintenance_preserves_symlink_targets(self): external = self.root/'external'; external.mkdir() external.joinpath('media').write_text('keep') self.releases.joinpath('e'*40).symlink_to(external) self.assertEqual(self.run_deploy('--maintenance').returncode, 0, self.output) self.assertTrue(external.joinpath('media').exists()) self.assert_old_live() def test_interrupted_build_recovered_by_maintenance(self): (self.state/'pending-deployment').write_text('building '+TARGET+'\n') (self.state/'previous-release').write_text(str(self.releases/PREVIOUS)) (self.releases/TARGET).mkdir() self.slots.joinpath('blue').unlink() self.slots.joinpath('blue').symlink_to(self.releases/TARGET) self.assertEqual(self.run_deploy('--maintenance').returncode, 0, self.output) self.assert_old_live() self.assertFalse((self.releases/TARGET).exists()) self.assertEqual(self.slots.joinpath('blue').resolve(), self.releases/PREVIOUS) def test_signal_cleans_build_process_and_candidate(self): with (self.root/'output').open('w') as output: process = subprocess.Popen(['bash', str(self.script), TARGET], env={**self.env, 'FAIL_STAGE': 'hang'}, stdout=output, stderr=output) try: deadline = time.monotonic() + 15 while not (self.root/'building').exists() and time.monotonic() < deadline: if process.poll() is not None: break time.sleep(.05) self.assertTrue((self.root/'building').exists(), (self.root/'output').read_text()) process.terminate() process.wait(timeout=15) finally: if process.poll() is None: process.kill(); process.wait() self.output = (self.root/'output').read_text() self.assertEqual(process.returncode, 143, self.output) self.assert_old_live() self.assertFalse((self.releases/TARGET).exists(), self.output) if __name__ == '__main__': unittest.main()