"""Synthetic deployment tests: no network, real services, credentials or real signals.""" from contextlib import nullcontext import importlib.util import json from pathlib import Path import signal import sqlite3 import tempfile import unittest from unittest.mock import Mock, patch SPEC = importlib.util.spec_from_file_location('deployment', Path(__file__).parents[1] / 'deploy.py') deployment = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(deployment) class DeployTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.d = deployment.Deployment(self.root) def controlled(self): (self.root / '.env').write_text('# synthetic; no credentials\n') self.d.lock = Mock(return_value=nullcontext()) self.d.processes = Mock(return_value=[{'pid': 123}]) for name in ('settings', 'stop', 'sync', 'models', 'update_knowledge', 'publish_knowledge', 'activate_runtime', 'start'): setattr(self.d, name, Mock()) parent = Mock() for name in ('stop', 'sync', 'models', 'update_knowledge', 'publish_knowledge', 'activate_runtime', 'start'): parent.attach_mock(getattr(self.d, name), name) return parent def test_first_deploy_installs_initializes_and_starts(self): parent = self.controlled() self.d.manifest_ready = Mock(return_value=False) self.assertEqual(self.d.execute('deploy'), 0) self.assertEqual([c[0] for c in parent.mock_calls], ['sync', 'models', 'update_knowledge', 'stop', 'publish_knowledge', 'activate_runtime', 'start']) def test_existing_deploy_does_not_regenerate_knowledge(self): parent = self.controlled() self.d.manifest_ready = Mock(return_value=True) self.d.execute('deploy') self.assertEqual([c[0] for c in parent.mock_calls], ['sync', 'models', 'stop', 'activate_runtime', 'start']) def test_independent_check_does_not_sync_or_stop_service(self): self.controlled() with patch('scripts.deploy_check.run', return_value=0) as run: self.assertEqual(self.d.execute('check-redis'), 0) run.assert_called_once_with(self.root, None, 'check-redis') self.d.sync.assert_not_called() self.d.stop.assert_not_called() def test_update_always_refreshes_and_restart_only_syncs_code(self): parent = self.controlled() self.d.execute('update-knowledge') self.assertEqual([c[0] for c in parent.mock_calls], ['sync', 'models', 'update_knowledge', 'stop', 'publish_knowledge', 'activate_runtime', 'start']) parent.reset_mock() self.d.execute('restart') self.assertEqual([c[0] for c in parent.mock_calls], ['sync', 'stop', 'activate_runtime', 'start']) def test_failures_do_not_start_service(self): for stage in ('stop', 'sync', 'models', 'update_knowledge'): with self.subTest(stage=stage): self.controlled() getattr(self.d, stage).side_effect = RuntimeError('synthetic_failure') with self.assertRaises(RuntimeError): self.d.execute('update-knowledge') self.d.start.assert_not_called() self.d.publish_knowledge.assert_not_called() self.d.activate_runtime.assert_not_called() if stage != 'stop': self.d.stop.assert_not_called() def test_missing_configuration_does_not_stop_service(self): self.controlled() (self.root / '.env').unlink() with self.assertRaisesRegex(RuntimeError, 'configure_dotenv'): self.d.execute('deploy') self.d.stop.assert_not_called() def test_knowledge_failure_preserves_live_snapshot(self): self.d.prepared.mkdir(parents=True) marker = self.d.prepared / 'old.txt' marker.write_text('previous') self.d.module = Mock(side_effect=RuntimeError('synthetic_failure')) with self.assertRaises(RuntimeError): self.d.update_knowledge() self.assertEqual(marker.read_text(), 'previous') def test_knowledge_pipeline_verifies_before_publish_and_copies_cache(self): self.d.prepared.mkdir(parents=True) (self.d.prepared / 'old.txt').write_text('previous') with sqlite3.connect(self.d.prepared / '.summary_cache.sqlite3') as db: db.execute('create table synthetic (value text)') db.execute("insert into synthetic values ('cached')") db.close() calls = [] def run(name, *args): calls.append((name, args)) stage = Path(args[1]) if name == 'src.step2_knowledge_ingest': (stage / 'ingestion').mkdir(exist_ok=True) (stage / 'ingestion/manifest.json').write_text('{"complete": true}') self.d.module = run stage = self.d.update_knowledge() # Even after remote validation, the old local snapshot stays live until stop. self.assertEqual((self.d.prepared / 'old.txt').read_text(), 'previous') self.d.publish_knowledge(stage) self.assertEqual(len(calls), 7) self.assertIn('--llm-workbook', calls[2][1]) self.assertIn('--verify-only', calls[3][1]) self.assertIn('--verify-only', calls[5][1]) self.assertIn('--verify', calls[6][1]) self.assertTrue(self.d.manifest_ready()) self.assertFalse((self.d.prepared / 'old.txt').exists()) self.assertEqual(len(list(self.d.output.glob('knowledge-*-previous/old.txt'))), 1) with sqlite3.connect(self.d.prepared / '.summary_cache.sqlite3') as db: self.assertEqual(db.execute('select value from synthetic').fetchone()[0], 'cached') db.close() def test_pid_reuse_never_signals(self): identity = {'pid': 123, 'start': 'old', 'boot': 'same'} self.d.identity = Mock(return_value={'pid': 123, 'start': 'new', 'boot': 'same'}) with patch.object(deployment.os, 'pidfd_open', create=True) as opened: self.d.send(identity, signal.SIGTERM) opened.assert_not_called() def test_identity_rechecked_after_pidfd_open(self): identity = {'pid': 123, 'start': 'old', 'boot': 'same'} self.d.identity = Mock(side_effect=[identity, None]) with patch.object(deployment.os, 'pidfd_open', return_value=88, create=True), \ patch.object(deployment.signal, 'pidfd_send_signal', create=True) as send, \ patch.object(deployment.os, 'close') as close: self.d.send(identity, signal.SIGTERM) send.assert_not_called() close.assert_called_once_with(88) def test_health_accepts_only_ready_and_disables_proxy(self): response = Mock(status=200) response.read.return_value = b'{"status":"ready"}' opener = Mock() opener.open.return_value.__enter__ = Mock(return_value=response) opener.open.return_value.__exit__ = Mock(return_value=False) with patch.object(deployment, 'build_opener', return_value=opener): self.assertTrue(self.d.healthy('::', 8000)) self.assertEqual(opener.open.call_args.args[0], 'http://[::1]:8000/health') response.read.return_value = b'{"status":"wrong"}' self.assertFalse(self.d.healthy('0.0.0.0', 8000)) def test_busy_port_does_not_launch_or_kill_another_service(self): self.d.settings = Mock(return_value=Mock(host='127.0.0.1', port=8000)) self.d.processes = Mock(return_value=[]) self.d.manifest_ready = Mock(return_value=True) with patch.object(deployment.socket, 'socket') as sock, \ patch.object(deployment.subprocess, 'Popen') as launch: sock.return_value.__enter__.return_value.bind.side_effect = OSError('busy') with self.assertRaises(OSError): self.d.start() launch.assert_not_called() def test_sync_uses_locked_dependencies_and_repo_interpreter(self): (self.root / 'uv.lock').write_text('synthetic') (self.root / '.python-version').write_text('3.14.5\n') self.d.run = Mock() self.d.sync() self.assertEqual(self.d.run.call_args_list[0].args, ('uv', 'sync', '--locked', '--python', '3.14.5')) self.assertEqual(self.d.run.call_args_list[1].args[0], self.d.python) target = self.d.run.call_args_list[0].kwargs['env']['UV_PROJECT_ENVIRONMENT'] self.assertEqual(Path(target), self.d.python.parent.parent) self.assertNotEqual(Path(target), self.root / '.venv') self.assertFalse(self.d.runtime_file.exists()) def test_offline_sync_and_models_never_request_download(self): (self.root / 'uv.lock').write_text('synthetic') (self.root / '.python-version').write_text('3.14.5\n') (self.root / '.venv/bin').mkdir(parents=True) (self.root / '.venv/bin/python').write_bytes(b'synthetic') self.d.run = Mock() with patch.dict(deployment.os.environ, {'DEPLOY_OFFLINE': '1'}): self.d.sync() self.assertEqual(len(self.d.run.call_args_list), 2) self.assertEqual(self.d.run.call_args_list[0].args[0], self.root / '.venv/bin/python') self.assertEqual(self.d.run.call_args_list[1].args[1], '-c') with patch('src.step2_knowledge_ingest.ingest.config', return_value={ 'EMBEDDING_MODEL_PATH': self.root / 'models/bge-large-zh-v1.5'}): self.d.models() self.assertIn('--check-only', self.d.run.call_args.args) def test_offline_deploy_imports_snapshot_before_stopping_service(self): parent = self.controlled() snapshot = self.root / 'output/knowledge-snapshot.zip' snapshot.write_bytes(b'synthetic') self.d.module = Mock() self.d.manifest_ready = Mock(side_effect=[False, True]) with patch.dict(deployment.os.environ, {'DEPLOY_OFFLINE': '1'}): self.assertEqual(self.d.execute('deploy'), 0) self.d.module.assert_called_once_with('src.step2_knowledge_ingest.portable', 'import', snapshot) self.assertNotIn('update_knowledge', [call[0] for call in parent.mock_calls]) self.assertEqual([call[0] for call in parent.mock_calls], ['sync', 'models', 'stop', 'activate_runtime', 'start']) def test_offline_deploy_requires_snapshot_without_stopping_service(self): self.controlled() self.d.manifest_ready = Mock(return_value=False) with patch.dict(deployment.os.environ, {'DEPLOY_OFFLINE': '1'}): with self.assertRaisesRegex(RuntimeError, 'offline_knowledge_snapshot_missing'): self.d.execute('deploy') self.d.stop.assert_not_called() def test_import_knowledge_is_independent_of_service_restart(self): self.controlled() snapshot = self.root / 'output/knowledge-snapshot.zip' snapshot.write_bytes(b'synthetic') self.d.run = Mock() self.d.manifest_ready = Mock(return_value=True) self.assertEqual(self.d.execute('import-knowledge'), 0) self.assertEqual(self.d.run.call_args.args[1:], ('-m', 'src.step2_knowledge_ingest.portable', 'import', snapshot)) self.d.sync.assert_not_called() self.d.stop.assert_not_called() self.d.start.assert_not_called() def test_import_custom_zip_and_library_arguments(self): self.controlled() snapshot = self.root / 'my knowledge.zip' snapshot.write_bytes(b'synthetic') self.d.run = Mock() self.d.manifest_ready = Mock(return_value=True) self.d.execute('import-knowledge', archive='my knowledge.zip', database='article') self.assertEqual(self.d.run.call_args.args[1:], ('-m', 'src.step2_knowledge_ingest.portable', 'import', snapshot, '--database', 'article')) self.d.execute('import-knowledge', archive=snapshot) self.assertEqual(self.d.run.call_args.args[1:], ('-m', 'src.step2_knowledge_ingest.portable', 'import', snapshot)) for action, archive, database in [('import-knowledge', snapshot, 'bad'), ('import-knowledge', None, 'article'), ('restart', snapshot, None)]: with self.assertRaises(RuntimeError): self.d.execute(action, archive=archive, database=database) self.d.stop.assert_not_called() self.d.start.assert_not_called() def test_import_cli_parameters_and_invalid_extra_arguments(self): with patch.object(deployment.sys, 'platform', 'linux'), \ patch.object(deployment, 'Deployment') as controller: controller.return_value.execute.return_value = 0 with patch.object(deployment.sys, 'argv', ['deploy.py', 'import-knowledge', '/some path/bank.zip', 'article']): self.assertEqual(deployment.main(), 0) controller.return_value.execute.assert_called_once_with( 'import-knowledge', archive='/some path/bank.zip', database='article') for args in (['restart','extra'], ['check-qcc'], ['import-knowledge','x.zip','wrong']): with self.subTest(args=args), patch.object(deployment.sys, 'argv', ['deploy.py', *args]), self.assertRaises(SystemExit): deployment.main() def test_health_timeout_stops_the_new_child_and_removes_pidfile(self): self.d.settings = Mock(return_value=Mock(host='127.0.0.1', port=8000)) self.d.processes = Mock(return_value=[]) self.d.manifest_ready = Mock(return_value=True) identity = {'pid': 123, 'start': 'start', 'boot': 'boot'} self.d.identity = Mock(return_value=identity) self.d.send = Mock() child = Mock(pid=123) child.poll.return_value = None with patch.object(deployment.socket, 'socket'), \ patch.object(deployment.subprocess, 'Popen', return_value=child), \ patch.dict(deployment.os.environ, DEPLOY_START_TIMEOUT='0'): with self.assertRaisesRegex(RuntimeError, 'api_health_timeout'): self.d.start() self.d.send.assert_called_once_with(identity, signal.SIGTERM) self.assertFalse(self.d.pidfile.exists()) def test_publication_failure_restores_old_directory(self): self.d.prepared.mkdir(parents=True) (self.d.prepared / 'old.txt').write_text('previous') def run(name, *args): stage = Path(args[1]) (stage / 'ingestion').mkdir(exist_ok=True) (stage / 'ingestion/manifest.json').write_text('{"complete":true}') self.d.module = run real_rename = Path.rename def rename(path, target): if path.name == 'knowledge-staging': raise OSError('synthetic_move_failure') return real_rename(path, target) stage = self.d.update_knowledge() with patch.object(Path, 'rename', rename): with self.assertRaisesRegex(OSError, 'synthetic_move_failure'): self.d.publish_knowledge(stage) self.assertEqual((self.d.prepared / 'old.txt').read_text(), 'previous') def test_active_service_models_are_read_only(self): self.controlled() self.d.execute('update-knowledge') self.d.models.assert_called_once_with(check_only=True) self.controlled() self.d.processes.return_value = [] self.d.execute('update-knowledge') self.d.models.assert_called_once_with(check_only=False) def test_runtime_pointer_changes_only_at_activation_and_survives_new_controller(self): legacy = self.d.python new = self.d.environments / 'env-synthetic/bin/python' self.d.python = new self.assertEqual(deployment.Deployment(self.root).python, legacy) self.d.activate_runtime() self.assertEqual(deployment.Deployment(self.root).python, new) self.assertTrue(self.d.owned_python(legacy)) self.assertTrue(self.d.owned_python(new)) self.assertFalse(self.d.owned_python(Path('/unrelated/bin/python'))) self.assertFalse(self.d.owned_python(self.root / 'output/deploy/environments/env-bad/../../foreign/bin/python')) def test_corrupt_runtime_pointer_rejected(self): self.d.runtime_file.write_text('/unrelated/bin/python\n') with self.assertRaisesRegex(RuntimeError, 'invalid_deployment_runtime_path'): deployment.Deployment(self.root) def test_runtime_smoke_accepts_explicit_staged_environment(self): environment = self.root / 'isolated environment' deployment.subprocess.run([deployment.sys.executable, '-m', 'venv', '--without-pip', str(environment)], check=True, capture_output=True) python = environment / ('Scripts/python.exe' if deployment.os.name == 'nt' else 'bin/python') script = Path(__file__).parents[1] / 'check-runtime.py' command = [str(python), str(script), '--environment', str(environment)] result = deployment.subprocess.run(command, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) result = deployment.subprocess.run([str(python), str(script)], capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) if __name__ == '__main__': unittest.main()