118 lines
5.5 KiB
Python
118 lines
5.5 KiB
Python
"""Deployment ordering checks using isolated fixtures and stubbed external commands."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
DOCKER_STUB = r"""#!/usr/bin/env python3
|
|
import json, os, sys
|
|
args = sys.argv[1:]
|
|
with open(os.environ['DEPLOY_TEST_TRACE'], 'a') as log:
|
|
log.write(json.dumps(args) + '\n')
|
|
if args[0] == 'inspect':
|
|
print('false' if '.State.Running' in args[2] else 'healthy')
|
|
elif args[0] == 'compose':
|
|
command = args[1:]
|
|
while command and command[0] == '-f':
|
|
command = command[2:]
|
|
if command[:2] == ['ps', '-q']:
|
|
if command[2] != 'laravel.test':
|
|
print('fixture-' + command[2])
|
|
if command and command[0] == 'exec' and 'sha256sum' in command:
|
|
print('fixture-hash /var/www/html/public/build/manifest.json')
|
|
if command and command[0] == 'exec' and '-r' in command:
|
|
print('/build/fixture.js')
|
|
print('/build/fixture.css')
|
|
if command and command[0] == 'exec' and any('SELECT 1' in value for value in command):
|
|
if os.environ.get('DEPLOY_TEST_DB_FAILURE') == '1':
|
|
sys.exit(17)
|
|
"""
|
|
|
|
|
|
class ProductionDeployTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix='moonwell-deploy-test-')
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
(self.root / 'scripts').mkdir()
|
|
(self.root / 'bin').mkdir()
|
|
shutil.copy2(REPO / 'scripts/deploy-production.sh', self.root / 'scripts')
|
|
for name in ['compose.prod.yaml', 'compose.prod.ssh-tunnel.yaml', 'compose.yaml']:
|
|
(self.root / name).touch()
|
|
self.trace = self.root / 'trace.jsonl'
|
|
for name, content in [('docker', DOCKER_STUB), ('curl', '#!/bin/sh\ncase "$*" in *fixture.js*) printf application/javascript ;; *fixture.css*) printf text/css ;; esac\n')]:
|
|
path = self.root / 'bin' / name
|
|
path.write_text(content)
|
|
path.chmod(0o755)
|
|
|
|
def run_deploy(self, content='', args=(), fail_db=False):
|
|
# Deliberately omit optional WWWUSER/WWWGROUP/APP_PORT settings.
|
|
(self.root / '.env').write_text(content)
|
|
env = dict(os.environ, PATH=str(self.root / 'bin') + ':' + os.environ['PATH'],
|
|
DEPLOY_TEST_TRACE=str(self.trace), DEPLOY_TEST_DB_FAILURE=str(int(fail_db)))
|
|
result = subprocess.run(['bash', str(self.root / 'scripts/deploy-production.sh'), *args],
|
|
cwd=self.root, env=env, capture_output=True, text=True)
|
|
calls = [json.loads(line) for line in self.trace.read_text().splitlines()] if self.trace.exists() else []
|
|
return result, calls
|
|
|
|
def operations(self, calls):
|
|
result = []
|
|
for call in calls:
|
|
if call[0] != 'compose':
|
|
continue
|
|
command = call[1:]
|
|
while command and command[0] == '-f':
|
|
command = command[2:]
|
|
result.append(command)
|
|
return result
|
|
|
|
def test_direct_database_mode_and_preflight_order(self):
|
|
result, calls = self.run_deploy('VMANGOS_AUTH_DB_HOST=192.0.2.1\n')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(any('compose.prod.ssh-tunnel.yaml' in arg for call in calls for arg in call))
|
|
ops = self.operations(calls)
|
|
probe = next(i for i, op in enumerate(ops) if any('SELECT 1' in arg for arg in op))
|
|
migration = next(i for i, op in enumerate(ops) if 'migrate' in op)
|
|
clear = next(i for i, op in enumerate(ops) if 'config:clear' in op)
|
|
self.assertLess(clear, probe)
|
|
self.assertLess(probe, migration)
|
|
|
|
def test_ssh_flag_starts_tunnel_before_app(self):
|
|
result, calls = self.run_deploy('VMANGOS_AUTH_DB_HOST=db\n', ['--ssh-tunnel'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
production_calls = [call for call in calls if any(arg.endswith('/compose.prod.yaml') for arg in call)]
|
|
self.assertTrue(production_calls)
|
|
self.assertTrue(all(any(arg.endswith('/compose.prod.ssh-tunnel.yaml') for arg in call)
|
|
for call in production_calls))
|
|
ops = self.operations(calls)
|
|
tunnel = ops.index(['up', '-d', '--no-deps', 'core-db-tunnel'])
|
|
app = ops.index(['up', '-d', '--no-deps', 'app'])
|
|
self.assertLess(tunnel, app)
|
|
self.assertTrue(all('--remove-orphans' not in call for call in calls))
|
|
|
|
def test_single_quoted_host_enables_ssh_automatically(self):
|
|
result, calls = self.run_deploy("VMANGOS_AUTH_DB_HOST='core-db-tunnel'\n")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(['build', 'core-db-tunnel'], self.operations(calls))
|
|
|
|
def test_failed_database_probe_prevents_migrations(self):
|
|
result, calls = self.run_deploy('VMANGOS_AUTH_DB_HOST=core-db-tunnel\n', fail_db=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
ops = self.operations(calls)
|
|
self.assertTrue(any(any('SELECT 1' in arg for arg in op) for op in ops))
|
|
self.assertFalse(any('migrate' in op or 'optimize' in op for op in ops))
|
|
|
|
def test_database_url_rejected_before_build(self):
|
|
result, calls = self.run_deploy('VMANGOS_AUTH_DB_HOST=core-db-tunnel\nVMANGOS_AUTH_DB_URL=mysql://fixture\n')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('VMANGOS_AUTH_DB_URL', result.stderr)
|
|
self.assertFalse(any(op and op[0] in ['build', 'up', 'exec'] for op in self.operations(calls)))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|