Corp/ops/deploy/tests/test_deploy_lifecycle.py

"""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 'config' in args:
        if '--get-regexp' in args:
            sys.exit(0 if os.environ.get('FAIL_STAGE') == 'url_rewrite' else 1)
        print('ssh://git@172.11.0.14/Robotail/Corp.git' if os.environ.get('FAIL_STAGE') != 'origin' else 'ssh://untrusted/other.git')
    elif '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 any('shutil.which' in a for a in args): fail('imagemagick')
    elif any('eusrr-deploy-healthcheck' in a for a in args):
        fail('websocket_candidate' if '--app-port' in args else 'websocket_transport')
    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:
        count_file=root/'npm_attempts'
        count=int(count_file.read_text())+1 if count_file.exists() else 1
        count_file.write_text(str(count))
        stage=os.environ.get('FAIL_STAGE')
        if stage=='npm_network_always' or (stage=='npm_network_once' and count==1):
            print('npm error code ECONNRESET', file=sys.stderr)
            sys.exit(152)
        if stage=='npm_dependency':
            print('npm error code ERESOLVE', file=sys.stderr)
            sys.exit(1)
        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 == 'node': fail('node_runtime')
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', 'node',
                     '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', 'imagemagick', 'websocket_transport', 'build', 'dump', 'migration', 'health',
                      'websocket_candidate', '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_transient_npm_network_failure_retries_same_candidate(self):
        result = self.run_deploy(FAIL_STAGE='npm_network_once')
        self.assertEqual(result.returncode, 0, self.output)
        self.assertEqual((self.root/'npm_attempts').read_text(), '2')
        self.assertIn('retrying the same lockfile', self.output)
        self.assertFalse((self.releases/TARGET/'.npm-install-attempt.log').exists())

    def test_persistent_npm_network_failure_is_bounded_and_cleans_candidate(self):
        result = self.run_deploy(FAIL_STAGE='npm_network_always')
        self.assertEqual(result.returncode, 152, self.output)
        self.assertEqual((self.root/'npm_attempts').read_text(), '3')
        self.assert_old_live()
        self.assertFalse((self.releases/TARGET).exists())

    def test_npm_dependency_errors_are_not_retried(self):
        result = self.run_deploy(FAIL_STAGE='npm_dependency')
        self.assertEqual(result.returncode, 1, self.output)
        self.assertEqual((self.root/'npm_attempts').read_text(), '1')
        self.assert_old_live()

    def test_unsupported_node_rejected_before_creating_candidate(self):
        result = self.run_deploy(FAIL_STAGE='node_runtime')
        self.assertEqual(result.returncode, 17, self.output)
        self.assert_old_live()
        self.assertFalse((self.releases/TARGET).exists())
        self.assertFalse((self.root/'npm_attempts').exists())

    def test_pip_bootstrap_is_pinned_and_download_resume_keeps_hash_checks(self):
        result = self.run_deploy()
        self.assertEqual(result.returncode, 0, self.output)
        commands = (self.root/'commands').read_text()
        self.assertIn('pip-bootstrap.txt', commands)
        self.assertIn('--require-hashes', commands)
        self.assertIn('--resume-retries=5', commands)
        self.assertNotIn('--trusted-host', commands)
        self.assertNotIn('--no-require-hashes', commands)
        bootstrap = (SOURCE.parent/'pip-bootstrap.txt').read_text()
        self.assertIn('pip==26.2.1 --hash=sha256:', bootstrap)

    def test_source_override_is_rejected_without_touching_live_release(self):
        for stage in ('origin', 'url_rewrite'):
            with self.subTest(stage=stage):
                self.assertEqual(self.run_deploy(FAIL_STAGE=stage).returncode, 5, self.output)
                self.assert_old_live()
                self.assertTrue((self.releases/PREVIOUS).exists())
                self.assertFalse((self.releases/TARGET).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_mutable_slot_control_directory_refuses_cleanup(self):
        self.slots.chmod(0o775)
        self.assertNotEqual(self.run_deploy('--maintenance').returncode, 0)
        self.assertTrue((self.releases/ORPHAN).exists())

    def test_slot_outside_release_root_refuses_cleanup(self):
        self.slots.joinpath('blue').unlink()
        self.slots.joinpath('blue').symlink_to(self.root/'srv/eusrr')
        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()