fix

tests/deployment/test_production_deploy.py

@@ -0,0 +1,117 @@
+"""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()