"""Focused local deployment regressions; Docker cases execute real containers.""" from pathlib import Path import subprocess import sys import tempfile import unittest from urllib.request import urlopen from local_app_install import release_directory from local_deployment import run, write, write_json, wait_http, worker_image_identity from local_l0 import current_serving_mode, l0_compose, wait_process_health from local_l1 import extract_frontend, l1_compose, nginx_config class LocalInputsTest(unittest.TestCase): def test_prepared_metadata_does_not_change_selected_runtime_mode(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) write_json(root / "l0/config/serving.json", {}) (root / "l0/config/workers").mkdir() for command, expected in ((["bootstrap"], False), (None, True), ([], True)): write_json(root / "l0/compose.json", {"services": {"runtime-api": {"command": command}}}) self.assertEqual(current_serving_mode(root), expected) write_json(root / "l0/compose.json", {"services": {"runtime-api": {"command": ["unknown"]}}}) with self.assertRaisesRegex(ValueError, "unknown local Runtime command"): current_serving_mode(root) def test_local_worker_identity_accepts_unlabeled_images(self): digest = 'sha256:' + 'a' * 64 for labels in (None, {}, {"org.opencontainers.image.revision": ""}): self.assertEqual(worker_image_identity('local-worker:dev', {"Id": digest, "Config": {"Labels": labels}}), (digest, "local")) self.assertEqual(worker_image_identity('local-worker:dev', {"Id": digest, "Config": {"Labels": {"org.opencontainers.image.revision": "working-tree"}}}), (digest, "working-tree")) def test_restart_cannot_change_selected_images(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) selected = root / 'l0/settings.json' write_json(selected, {'runtime_image': 'original', 'worker_image': 'original-worker'}) original = selected.read_bytes() for command in ('activate', 'start'): for option in ('--runtime-image', '--worker-image'): result = subprocess.run([sys.executable, str(Path(__file__).with_name('local_l0.py')), command, '--root', directory, option, 'replacement'], capture_output=True, text=True) self.assertEqual(result.returncode, 2, result.stderr) self.assertIn('only available during init', result.stderr) self.assertEqual(selected.read_bytes(), original) def test_separate_projects_share_container_network(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) settings = {'project': 'test-local', 'runtime_image': 'runtime-selected', 'worker_image': 'worker-selected', 'grpc_port': 3312, 'admin_port': 3313, 'app_image': 'app-selected', 'app_port': 3302, 'public_port': 3300, 'frontend_path': str(root / 'frontend')} l0 = l0_compose(root, settings, False) l1 = l1_compose(root, settings) self.assertNotEqual(l0['name'], l1['name']) for config in (l0, l1): self.assertEqual(config['networks']['default'], {'name': 'test-local-backend', 'external': True}) for service in config['services'].values(): self.assertNotIn('extra_hosts', service) for port in service.get('ports', []): self.assertTrue(port.startswith('127.0.0.1:')) self.assertEqual(l0['services']['runtime-api']['image'], 'runtime-selected') def test_release_path_stays_inside_checkout(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) checkout = root / 'checkout' checkout.mkdir() (checkout / 'outside').symlink_to(root, target_is_directory=True) self.assertEqual(release_directory(checkout, 'release/app-package.json'), checkout.resolve() / 'release') for path in ('../app-package.json', str(root / 'app-package.json'), 'outside/app-package.json'): with self.assertRaises(ValueError): release_directory(checkout, path) class LocalHealthTest(unittest.TestCase): def test_api_without_inherited_healthcheck_uses_the_http_readiness_owner(self): expected = {"runtime-api": {}, "worker-scheduled-flow": {"healthcheck": {}}} observed = [ {"Config": {"Labels": {"com.docker.compose.service": "runtime-api"}}, "State": {"Running": True}}, {"Config": {"Labels": {"com.docker.compose.service": "worker-scheduled-flow"}}, "State": {"Running": True, "Health": {"Status": "healthy"}}}, ] wait_process_health(["api", "worker"], expected, inspect=lambda: observed) def test_transient_unhealthy_restart_must_recover_before_success(self): elapsed = [0] calls = [] def inspect(): calls.append(elapsed[0]) ready = elapsed[0] >= 4 return [{"Config": {"Labels": {"com.docker.compose.service": "worker-scheduled-flow"}}, "State": {"Running": ready, "Health": {"Status": "healthy" if ready else "unhealthy"}}}] wait_process_health(["id"], {"worker-scheduled-flow": {"healthcheck": {}}}, inspect=inspect, timeout=10, clock=lambda: elapsed[0], pause=lambda seconds: elapsed.__setitem__(0, elapsed[0] + seconds)) self.assertEqual(calls, [0, 2, 4]) def test_persistent_failure_reports_bounded_pending_service(self): elapsed = [0] def inspect(): return [{"Config": {"Labels": {"com.docker.compose.service": "worker-scheduled-flow"}}, "State": {"Running": False, "Status": "restarting"}}] with self.assertRaisesRegex(RuntimeError, 'worker-scheduled-flow.*restarting'): wait_process_health(["id"], {"worker-scheduled-flow": {"healthcheck": {}}}, inspect=inspect, timeout=4, clock=lambda: elapsed[0], pause=lambda seconds: elapsed.__setitem__(0, elapsed[0] + seconds)) self.assertEqual(elapsed[0], 4) def test_missing_process_is_not_a_healthy_empty_inventory(self): with self.assertRaisesRegex(RuntimeError, 'inventory'): wait_process_health(["id"], {"worker-scheduled-flow": {"healthcheck": {}}}, inspect=lambda: []) class DockerDeploymentTest(unittest.TestCase): def test_proxy_and_frontend_replacement(self): with tempfile.TemporaryDirectory(prefix='local-deployment-test-') as directory: root = Path(directory) project = root.name network = project + '-backend' image = project + ':assets' upstream = project + '-upstream' frontend = project + '-frontend' try: run(['docker', 'network', 'create', network], capture=True) write(root / 'upstream.conf', 'server { listen 8080; location / { return 200 "backend:$uri:$http_upgrade"; } }', 0o644) run(['docker', 'run', '-d', '--name', upstream, '--network', network, '--network-alias', 'app-service', '-v', f'{root}/upstream.conf:/etc/nginx/conf.d/default.conf:ro', 'nginx:1.27.5-alpine'], capture=True) write(root / 'Dockerfile', 'FROM nginx:1.27.5-alpine\nCOPY assets/ /usr/share/nexuscrm/frontend/\n') write(root / 'assets/index.html', 'first release', 0o644) write(root / 'assets/removed.js', 'obsolete', 0o644) run(['docker', 'build', '-q', '-t', image, root], capture=True) old = extract_frontend(image, root) (root / 'assets/removed.js').unlink() write(root / 'assets/index.html', 'second release', 0o644) run(['docker', 'build', '-q', '-t', image, root], capture=True) current = extract_frontend(image, root) self.assertTrue((old / 'removed.js').exists()) self.assertFalse((current / 'removed.js').exists()) self.assertEqual((current / 'index.html').read_text(), 'second release') write(root / 'nginx.conf', nginx_config(), 0o644) run(['docker', 'run', '-d', '--name', frontend, '--network', network, '-p', '127.0.0.1::80', '-v', f'{root}/nginx.conf:/etc/nginx/conf.d/default.conf:ro', '-v', f'{current}:/usr/share/nginx/html:ro', 'nginx:1.27.5-alpine'], capture=True) port = run(['docker', 'port', frontend, '80'], capture=True).strip() origin = 'http://' + port wait_http(origin) with urlopen(origin) as response: self.assertEqual(response.read(), b'second release') for path in ('services/data/v63.0', 'api/test', 'oauth/test', 'oauth2/authorize', 'product/test', 'ws', '.well-known/oauth-authorization-server', 'downloads/test', 'mcp', 'readyz', 'livez'): with urlopen(origin + '/' + path) as response: self.assertEqual(response.read().decode(), 'backend:/' + path + ':') headers = run(['docker', 'exec', frontend, 'wget', '-qO-', '--header=Upgrade: websocket', '--header=Connection: Upgrade', 'http://127.0.0.1/ws'], capture=True) self.assertEqual(headers, 'backend:/ws:websocket') finally: for name in (frontend, upstream): subprocess.run(['docker', 'rm', '-f', name], capture_output=True) subprocess.run(['docker', 'image', 'rm', image], capture_output=True) subprocess.run(['docker', 'network', 'rm', network], capture_output=True) if __name__ == '__main__': unittest.main()