qtalker---/backend/tests/conftest.py

import os

import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.config import settings
from app.db import Base, get_db
from app.main import app

def _test_database_url() -> str:
    """A SEPARATE database from the configured one.

    This is not a nicety. The autouse fixture below drop_all()s every table
    before each test, so pointing it at settings.database_url means running
    the suite silently annihilates the live install — every account, every
    discovered spirit, every Ghost Log. That is exactly what used to happen:
    a full test run left the production Codex at zero entities.

    Honouring TEST_DATABASE_URL when set (for CI), and otherwise deriving
    `<configured-db>_test`, so the suite can never touch real data even if
    somebody runs it on the production host — which is the normal case here,
    since this box both serves the app and holds the repo.
    """
    explicit = os.environ.get("TEST_DATABASE_URL")
    if explicit:
        return explicit
    base, _, name = settings.database_url.rpartition("/")
    if not name:
        raise RuntimeError(
            "cannot derive a test database from DATABASE_URL; "
            "set TEST_DATABASE_URL explicitly"
        )
    # Strip any query string (e.g. ?ssl=require) before suffixing the name.
    db_name, sep, query = name.partition("?")
    return f"{base}/{db_name}_test{sep}{query}"


TEST_DATABASE_URL = _test_database_url()

# Fail loudly rather than eating the live data if the guard above is ever
# defeated by an unusual URL shape.
if TEST_DATABASE_URL == settings.database_url:
    raise RuntimeError(
        "refusing to run: the test database resolved to the production "
        "database, and the suite drops every table"
    )

# pytest-asyncio gives each test function its own event loop by default;
# asyncpg connections are bound to the loop they were opened on, so a pooled
# connection from one test's loop breaks the next test. NullPool sidesteps
# this by opening a fresh connection per checkout — scoped to this dedicated
# test engine so app.db.engine (used by production) is unaffected.
test_engine = create_async_engine(TEST_DATABASE_URL, poolclass=NullPool)
TestSessionLocal = async_sessionmaker(test_engine, expire_on_commit=False)


@pytest_asyncio.fixture(autouse=True)
async def _reset_db():
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    yield


async def _override_get_db():
    async with TestSessionLocal() as session:
        yield session


app.dependency_overrides[get_db] = _override_get_db


@pytest_asyncio.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="https://test") as ac:
        yield ac


@pytest.fixture
def sync_client(monkeypatch):
    """Sync TestClient (supports WebSocket tests). The WS handler's session
    maker is swapped to the NullPool test engine so its writes land in the
    same test database the async fixtures see. The lifespan's engine is also
    swapped: TestClient runs the lifespan on its own portal loop, and the
    pooled production engine would carry connections across loops."""
    import app.main as main_module
    import app.routes.device as device_module
    import app.ws as ws_module

    monkeypatch.setattr(ws_module, "session_maker", TestSessionLocal)
    monkeypatch.setattr(device_module, "session_maker", TestSessionLocal)
    monkeypatch.setattr(main_module, "engine", test_engine)
    with TestClient(app, base_url="https://testserver") as tc:
        yield tc


@pytest_asyncio.fixture
async def db_session():
    async with TestSessionLocal() as session:
        yield session