business-assistant/scripts/tests/test_deploy.py

"""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_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()