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


def _database_name(url: str) -> str:
    """Bare database name from a SQLAlchemy URL, sans query string."""
    return url.rpartition("/")[2].partition("?")[0]


# Two independent guards, because the suite drop_all()s every table before
# every single test and this box both serves the live app and holds the repo.
# Getting this wrong once already cost a production Codex.
#
# 1. Not literally the configured URL.
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"
    )

# 2. The database NAME must end in `_test`. A pure URL comparison is not
#    enough — `postgresql+asyncpg://u:p@localhost/quantumancy` and
#    `...@127.0.0.1/quantumancy` are different strings addressing the same
#    database, so guard #1 alone would happily wipe it. Requiring the name
#    itself to be a test database cannot be defeated by how the host is
#    spelled, and it also catches a TEST_DATABASE_URL that was set by hand
#    to something live.
_test_db_name = _database_name(TEST_DATABASE_URL)
if not _test_db_name.endswith("_test"):
    raise RuntimeError(
        f"refusing to run: test database {_test_db_name!r} does not end in "
        "'_test'. The suite drops every table; it must never be pointed at a "
        "database that could hold real seekers, spirits or Ghost Logs."
    )

# 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