LangBot/tests/e2e/test_startup.py

"""E2E tests for LangBot startup flow.

Tests the complete startup process including:
- boot.py startup orchestration
- stages/ (build_app, load_config, migrate, etc.)
- database initialization
- API availability

Run: uv run pytest tests/e2e/test_startup.py -v -m e2e
"""

from __future__ import annotations

import pytest

pytestmark = pytest.mark.e2e


@pytest.fixture(scope='session')
def e2e_tmpdir(tmp_path_factory):
    # Initial registration is global; keep this module independent of other E2E accounts.
    return tmp_path_factory.mktemp('langbot_startup')


@pytest.fixture(scope='session')
def e2e_port():
    import socket

    with socket.socket() as listener:
        listener.bind(('127.0.0.1', 0))
        return listener.getsockname()[1]


@pytest.fixture(scope='session')
def langbot_process(e2e_tmpdir, e2e_port):
    from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories
    from tests.e2e.utils.process_manager import LangBotProcess, find_project_root

    create_minimal_config(e2e_tmpdir, port=e2e_port)
    create_test_directories(e2e_tmpdir)
    process = LangBotProcess(
        project_root=find_project_root(),
        work_dir=e2e_tmpdir,
        port=e2e_port,
        timeout=180,
        collect_coverage=True,
    )
    try:
        assert process.start(), process.get_logs()
        yield process
    finally:
        process.stop()
        coverage_file = process.get_coverage_file()
        if coverage_file is not None and coverage_file.exists():
            import shutil

            shutil.copy(coverage_file, find_project_root() / '.coverage.e2e.startup')


@pytest.fixture(scope='session')
def e2e_db_path(e2e_tmpdir):
    return e2e_tmpdir / 'data/langbot.db'


@pytest.fixture
def startup_credentials(e2e_client):
    credentials = {'user': '[email protected]', 'password': 'startup-local-test-password'}
    state = e2e_client.get('/api/v1/user/init')
    assert state.status_code == 200
    if not state.json()['data']['initialized']:
        response = e2e_client.post('/api/v1/user/init', json=credentials)
        assert response.status_code == 200, response.text
        assert response.json()['code'] == 0
    return credentials


@pytest.fixture
def startup_headers(e2e_client, startup_credentials):
    response = e2e_client.post('/api/v1/user/auth', json=startup_credentials)
    assert response.status_code == 200, response.text
    headers = {'Authorization': f'Bearer {response.json()["data"]["token"]}'}
    bootstrap = e2e_client.get('/api/v1/workspaces/bootstrap', headers=headers)
    assert bootstrap.status_code == 200, bootstrap.text
    headers['X-Workspace-Id'] = bootstrap.json()['data']['workspaces'][0]['workspace']['uuid']
    return headers


class TestStartupFlow:
    """Tests for LangBot startup process."""

    def test_process_is_running(self, langbot_process):
        """Verify LangBot process is running."""
        assert langbot_process.is_running()

    def test_health_check(self, langbot_process, e2e_port):
        """Verify LangBot API is responding."""
        assert langbot_process.health_check()

    def test_health_check_exposes_bounded_blocking_executor(self, e2e_client):
        """The production startup path installs blocking-work admission."""
        response = e2e_client.get('/healthz')

        assert response.status_code == 200
        executor = response.json()['resources']['blocking_executor']
        assert executor['max_workers'] == 8
        assert executor['max_pending'] == 128
        assert executor['max_inflight_per_scope'] == 4
        assert executor['inflight'] >= 0
        assert executor['rejected_total'] >= 0
        event_loop = response.json()['resources']['event_loop']
        assert event_loop['running'] is True
        assert event_loop['recent_max_lag_ms'] >= 0

    def test_system_info_endpoint(self, e2e_client):
        """Test /api/v1/system/info endpoint."""
        response = e2e_client.get('/api/v1/system/info')
        assert response.status_code == 200

        data = response.json()
        assert data['code'] == 0
        assert 'data' in data
        # System info should contain version info
        assert 'version' in data['data'] or 'edition' in data['data']

    def test_database_initialized(self, langbot_process, e2e_db_path):
        """Verify SQLite database was created and initialized."""
        assert e2e_db_path.exists()

        # Database should have some tables after migration
        import sqlite3

        conn = sqlite3.connect(str(e2e_db_path))
        cursor = conn.cursor()

        # Check that core tables exist
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
        tables = [row[0] for row in cursor.fetchall()]

        # Core tables should be created by Alembic migrations
        # Note: table names may differ (legacy_pipelines instead of pipelines)
        expected_tables = ['legacy_pipelines', 'bots', 'model_providers', 'llm_models']
        for table in expected_tables:
            assert table in tables, f'Table {table} should exist. Available: {tables}'

        conn.close()

    def test_chroma_directory_created(self, e2e_tmpdir):
        """Verify Chroma vector database directory was created."""
        chroma_path = e2e_tmpdir / 'chroma'
        # Created by the E2E config factory before startup.
        assert chroma_path.exists()

    def test_pipelines_endpoint(self, e2e_client):
        """Test /api/v1/pipelines endpoint (requires auth)."""
        # Without auth, should return 401
        response = e2e_client.get('/api/v1/pipelines')
        assert response.status_code == 401

    def test_auth_endpoint(self, e2e_client, startup_credentials):
        """A controlled local account can sign in and use its token."""
        response = e2e_client.post('/api/v1/user/auth', json=startup_credentials)
        assert response.status_code == 200, response.text
        assert response.json()['code'] == 0
        token = response.json()['data']['token']
        assert isinstance(token, str) and token
        bootstrap = e2e_client.get('/api/v1/workspaces/bootstrap', headers={'Authorization': f'Bearer {token}'})
        assert bootstrap.status_code == 200, bootstrap.text
        assert bootstrap.json()['data']['workspaces'][0]['membership']['role'] == 'owner'

        rejected = e2e_client.post('/api/v1/user/auth', json={**startup_credentials, 'password': 'wrong-password'})
        assert rejected.status_code == 200, rejected.text
        assert rejected.json()['code'] == 1
        assert rejected.json().get('data', {}).get('token') is None


class TestStartupStages:
    """Tests that verify individual startup stages worked correctly."""

    def test_config_loaded(self, e2e_client):
        """Verify config was loaded correctly by checking API port."""
        # If API responds on e2e_port, config was loaded
        assert e2e_client.get('/api/v1/system/info').status_code == 200

    def test_migrations_applied(self, langbot_process, e2e_db_path):
        """Verify database migrations were applied."""
        import sqlite3

        conn = sqlite3.connect(str(e2e_db_path))
        cursor = conn.cursor()

        # Check alembic_version table exists and has version
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='alembic_version';")
        result = cursor.fetchone()
        assert result is not None, 'alembic_version table should exist'

        cursor.execute('SELECT version_num FROM alembic_version;')
        version = cursor.fetchone()
        from pathlib import Path

        from alembic.script import ScriptDirectory

        migrations = Path(__file__).parents[2] / 'src/langbot/pkg/persistence/alembic'
        assert version == (ScriptDirectory(str(migrations)).get_current_head(),)

        conn.close()

    def test_http_controller_initialized(self, e2e_client, startup_headers):
        """Authenticated business endpoints return their documented collections."""
        info = e2e_client.get('/api/v1/system/info', headers=startup_headers)
        assert info.status_code == 200
        assert isinstance(info.json()['data']['version'], str)

        for endpoint, key in [
            ('/api/v1/pipelines', 'pipelines'),
            ('/api/v1/provider/providers', 'providers'),
            ('/api/v1/platform/bots', 'bots'),
        ]:
            response = e2e_client.get(endpoint, headers=startup_headers)
            assert response.status_code == 200, (endpoint, response.text)
            payload = response.json()
            assert payload['code'] == 0
            assert isinstance(payload['data'][key], list)
            if key == 'providers':
                assert [(item['name'], item['requester'], item['api_keys']) for item in payload['data'][key]] == [
                    ('LangBot Models', 'space-chat-completions', [])
                ]
            else:
                assert payload['data'][key] == []


class TestMinimalStartupNoLLM:
    """Tests verifying LangBot can start without LLM providers."""

    def test_api_available_without_llm(self, e2e_client):
        """API should be available even without LLM providers configured."""
        response = e2e_client.get('/api/v1/system/info')
        assert response.status_code == 200

    def test_pipeline_metadata_available(self, e2e_client, startup_headers):
        """All pipeline configuration groups remain readable without an LLM."""
        response = e2e_client.get('/api/v1/pipelines/_/metadata', headers=startup_headers)
        assert response.status_code == 200, response.text
        payload = response.json()
        assert payload['code'] == 0
        configs = payload['data']['configs']
        assert [config['name'] for config in configs] == ['trigger', 'safety', 'ai', 'speech', 'output']
        ai = configs[2]
        runner = next(stage for stage in ai['stages'] if stage['name'] == 'runner')
        assert any(item['name'] == 'id' for item in runner['config'])