language-llm/backend/tests/test_startup.py

"""FluentGerman.ai — Startup / lifespan tests.

The API tests drive the app through httpx's ASGITransport, which does NOT run
lifespan events — so table creation and the admin bootstrap in main.lifespan
were never exercised. The suite could stay green while the app failed to boot.
"""

import asyncio

import pytest
from sqlalchemy import func, inspect as sa_inspect, select
from sqlalchemy.exc import IntegrityError, OperationalError

from app import main
from app.config import get_settings
from app.database import Base
from app.models import User
from tests.conftest import test_engine as engine, test_session as session_factory


@pytest.fixture
def startup_app(monkeypatch):
    """Point main's module-level engine/session at the test database."""
    monkeypatch.setattr(main, "engine", engine)
    monkeypatch.setattr(main, "async_session", session_factory)
    return main.app


@pytest.mark.asyncio
async def test_lifespan_creates_schema_and_bootstraps_admin(startup_app):
    """Starting from an empty database, startup must create tables and an admin."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    async with main.lifespan(startup_app):
        pass

    async with session_factory() as db:
        result = await db.execute(select(User).where(User.is_admin.is_(True)))
        admin = result.scalar_one_or_none()

    assert admin is not None, "startup did not create an admin user"
    assert admin.username == get_settings().admin_username
    assert admin.hashed_password != get_settings().admin_password, "password stored in clear"


@pytest.mark.asyncio
async def test_lifespan_is_idempotent(startup_app):
    """A restart must not create a second admin."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    for _ in range(2):
        async with main.lifespan(startup_app):
            pass

    async with session_factory() as db:
        result = await db.execute(
            select(func.count()).select_from(User).where(User.is_admin.is_(True))
        )
        assert result.scalar_one() == 1


@pytest.mark.asyncio
async def test_lifespan_survives_concurrent_workers(startup_app):
    """deploy/fluentgerman.service starts uvicorn with --workers 2, and each
    worker runs this lifespan. On an empty database they race on the same
    check-then-insert; neither worker may fail."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    async def boot():
        async with main.lifespan(startup_app):
            pass

    await asyncio.gather(boot(), boot())

    async with session_factory() as db:
        result = await db.execute(
            select(func.count()).select_from(User).where(User.is_admin.is_(True))
        )
        assert result.scalar_one() == 1, "concurrent startup created a duplicate admin"


@pytest.mark.asyncio
async def test_lifespan_reraises_real_ddl_failures(startup_app, monkeypatch):
    """Only a genuine 'table already exists' race may be swallowed — a
    permission or connection failure has to stop the worker."""
    def explode(*args, **kwargs):
        raise OperationalError(
            "CREATE TABLE users (...)", {},
            Exception("(1142, \"CREATE command denied to user 'fluentgerman'\")"),
        )

    monkeypatch.setattr(main.Base.metadata, "create_all", explode)

    with pytest.raises(OperationalError):
        async with main.lifespan(startup_app):
            pass


@pytest.mark.asyncio
async def test_lifespan_refuses_to_start_on_incomplete_schema(startup_app, monkeypatch):
    """If some tables are missing, starting anyway would leave the app broken
    in a way the admin query alone would not notice."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    monkeypatch.setattr(main, "SCHEMA_RETRY_DELAY", 0)
    monkeypatch.setattr(main.Base.metadata, "create_all", lambda *a, **k: None)

    with pytest.raises(RuntimeError, match="schema is incomplete"):
        async with main.lifespan(startup_app):
            pass


@pytest.mark.asyncio
async def test_lifespan_waits_out_a_partial_schema_race(startup_app, monkeypatch):
    """An 'already exists' error aborts create_all part-way, so the losing
    worker can briefly see a schema the winner is still finishing. That is a
    moment to retry, not a reason to refuse to start."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    real_create_all = main.Base.metadata.create_all
    calls = {"n": 0}

    def racing_create_all(*args, **kwargs):
        calls["n"] += 1
        if calls["n"] == 1:
            # the winner got there first and we bailed out having created nothing
            raise OperationalError(
                "CREATE TABLE users (...)", {},
                Exception("table 'users' already exists"),
            )
        return real_create_all(*args, **kwargs)

    monkeypatch.setattr(main, "SCHEMA_RETRY_DELAY", 0)
    monkeypatch.setattr(main.Base.metadata, "create_all", racing_create_all)

    async with main.lifespan(startup_app):
        pass

    assert calls["n"] >= 2, "should have retried after the race"

    async with engine.connect() as conn:
        present = await conn.run_sync(lambda c: set(sa_inspect(c).get_table_names()))
    assert set(Base.metadata.tables) <= present


@pytest.mark.asyncio
async def test_lifespan_fails_loudly_when_admin_name_is_taken(startup_app):
    """A non-admin already holding the configured username is not a lost race —
    starting with no administrator at all would be worse than refusing."""
    async with session_factory() as db:
        db.add(User(
            username=get_settings().admin_username,
            email="someone-else@example.com",
            hashed_password="x",
            is_admin=False,
        ))
        await db.commit()

    with pytest.raises(IntegrityError):
        async with main.lifespan(startup_app):
            pass