"""Run the real shell/helper in a disposable checkout with fake Git/Docker commands.""" import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import preprod_config class DeploymentOrderingTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.repo = self.root / 'pingu-concerts' (self.repo / 'scripts').mkdir(parents=True) self.env = self.root / 'preprod.env' self.secret = self.root / 'firebase.json' self.secret.write_text('synthetic test bytes, not a credential') self.secret.chmod(0o600) self.values = {name: 'synthetic-private-value' for name in preprod_config.REQUIRED} self.values.update(PUSH_ENABLED='true', COOKIE_SECURE='true', FIREBASE_PROJECT_ID='metalcircle-30d9b', FIREBASE_SERVICE_ACCOUNT_FILE=str(self.secret)) self.write_env() source = Path(__file__).resolve().parents[1] script = (source / 'deploy-preprod.sh').read_text().replace( "ENV_FILE='/home/kai/.config/metalcircle/preprod.env'", 'ENV_FILE=' + repr(str(self.env))) (self.repo / 'scripts/deploy-preprod.sh').write_text(script) shutil.copy(source / 'preprod_config.py', self.repo / 'scripts') self.control = self.root / 'control.json' self.control.write_text('{}') self.trace = self.root / 'trace.jsonl' self.bin = self.root / 'bin' self.bin.mkdir() shared = (f'CONTROL = {str(self.control)!r}\nTRACE = {str(self.trace)!r}\n' 'import json, sys\nfrom pathlib import Path\n' 'control = json.loads(Path(CONTROL).read_text())\n' 'args = sys.argv[1:]\n' 'with open(TRACE, "a") as trace: trace.write(json.dumps([Path(sys.argv[0]).name, *args]) + "\\n")\n') self.executable('git', shared + f''' if args == ['rev-parse', '--show-toplevel']: print({str(self.repo)!r}) elif args == ['branch', '--show-current']: print(control.get('branch', 'main')) elif args[0] == 'status': if control.get('fail') == 'status': sys.exit(1) print(control.get('dirty', '')) elif args[0] == 'log': print('abc123 synthetic test commit') elif args[0] == 'pull' and control.get('fail') == 'pull': sys.exit(1) ''') self.executable('sudo', shared + ''' assert args[:3] == ['docker', 'compose', '--env-file'] env_file = Path(args[3]) assert args[4:8] == ['-f', 'compose.yml', '-f', 'compose.preprod.yml'] action = args[8:] step = action[0] if step in ('run', 'exec'): step += '-firebase' if 'push_preflight.py' in action else '-gitea' if control.get('no_pass') == step: print('synthetic-private-value') sys.exit(0) if control.get('fail') == step: print('FAIL: gitea_http_403' if 'gitea' in step else 'FAIL: credential_service_account_mismatch') print('synthetic-private-value', file=sys.stderr) sys.exit(1) if action == ['config', '--no-interpolate', '--environment']: print(env_file.read_text()) elif action[0] in ('run', 'exec'): print('PASS: synthetic preflight') if control.get('edit') and step == 'run-gitea': env_file.write_text(env_file.read_text() + '\\nCHANGED=true\\n') elif action[0] == 'logs': print('private synthetic-private-value') print('web | INFO \\"POST /private/synthetic-private-value HTTP/1.1\\" 200 OK') elif action[0] == 'ps': print(json.dumps([{'Name': 'pingu-concerts-web', 'State': 'running'}])) ''') def executable(self, name, code): path = self.bin / name path.write_text('#!/usr/bin/env python3\n' + code) path.chmod(0o700) def write_env(self): self.env.write_text('\n'.join(k + '=' + v for k, v in self.values.items())) self.env.chmod(0o600) def run_deploy(self, *args, **controls): self.control.write_text(json.dumps(controls)) self.trace.write_text('') result = subprocess.run(['bash', str(self.repo / 'scripts/deploy-preprod.sh'), *args], cwd=self.root, env={**os.environ, 'PATH': str(self.bin) + ':' + os.environ['PATH']}, capture_output=True, text=True) self.assertNotIn('synthetic-private-value', result.stdout + result.stderr) commands = [json.loads(line) for line in self.trace.read_text().splitlines()] self.actions = [cmd[9:] for cmd in commands if cmd[0] == 'sudo'] return result def assert_no_update(self, result): self.assertNotEqual(result.returncode, 0) self.assertFalse(any(action[0] == 'up' for action in self.actions)) def test_success_order_and_non_destructive_commands(self): result = self.run_deploy() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) update = self.actions.index(['up', '-d', '--no-deps', 'web']) preflights = [a for a in self.actions[:update] if a[0] == 'run'] self.assertEqual(len(preflights), 2) self.assertTrue(all('--no-deps' in a and '--rm' in a for a in preflights)) self.assertIn('push_preflight.py', preflights[0]) self.assertIn('gitea_preflight.py', preflights[1]) self.assertEqual(len([a for a in self.actions[update + 1:] if a[0] == 'exec']), 2) self.assertEqual(len(list((self.root / 'backups').glob('preprod.env.*'))), 1) self.assertTrue(all(not set(a) & {'down', 'rm', '--force-recreate', 'db'} for a in self.actions)) def test_missing_env_secret_or_any_gitea_value_never_updates(self): for missing in ('GITEA_URL', 'GITEA_TOKEN', 'GITEA_OWNER', 'GITEA_REPO'): with self.subTest(missing=missing): saved = self.values.pop(missing) self.write_env() result = self.run_deploy() self.assert_no_update(result) self.assertIn('ERROR: required variable ' + missing + ' is missing', result.stdout) self.assertIn('Running container unchanged.', result.stderr) self.values[missing] = saved self.write_env() self.env.unlink() self.assert_no_update(self.run_deploy()) self.write_env() self.secret.unlink() self.assert_no_update(self.run_deploy()) def test_every_failure_before_up_keeps_running_container(self): for fail in ('status', 'pull', 'config', 'build', 'run-firebase', 'run-gitea'): with self.subTest(fail=fail): result = self.run_deploy(fail=fail) self.assert_no_update(result) self.assertIn('Running container unchanged.', result.stderr) self.assertFalse((self.root / 'backups').exists()) def test_dirty_tree_and_wrong_branch_abort_without_compose(self): for controls in ({'branch': 'feature'}, {'dirty': ' M changed-file'}): result = self.run_deploy(**controls) self.assert_no_update(result) self.assertEqual(self.actions, []) def test_check_only_never_updates_or_backs_up(self): result = self.run_deploy('--check') self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertFalse(any(a[0] in ('up', 'exec') for a in self.actions)) self.assertFalse((self.root / 'backups').exists()) def test_config_edit_during_preflight_aborts(self): result = self.run_deploy(edit=True) self.assert_no_update(result) self.assertIn('changed during deployment', result.stderr) def test_zero_exit_without_pass_is_not_sufficient(self): for step in ('run-firebase', 'run-gitea'): with self.subTest(step=step): self.assert_no_update(self.run_deploy(no_pass=step)) def test_backup_failure_aborts_before_update(self): (self.root / 'backups').mkdir(mode=0o755) self.assert_no_update(self.run_deploy()) def test_post_update_failure_does_not_claim_old_container_unchanged(self): result = self.run_deploy(fail='exec-gitea') self.assertNotEqual(result.returncode, 0) self.assertIn('Manual investigation required.', result.stderr) self.assertNotIn('Running container unchanged.', result.stderr) self.assertEqual(len([a for a in self.actions if a[0] == 'up']), 1)