hunjing/backend/tests/conftest.py

"""测试环境配置。

数据库支持两种模式：
- SQLite（默认）：自动创建临时文件库，每次测试 drop_all/create_all 隔离
- PostgreSQL：设环境变量 ``HUNJING_TEST_DATABASE_URL``，指向测试数据库
  （名称须含 ``test``），例如：
  ``HUNJING_TEST_DATABASE_URL=postgresql+asyncpg://hunjing:change-me@127.0.0.1:5432/hunjing_test``
"""

import os
import tempfile
from collections.abc import AsyncGenerator, AsyncIterator
from pathlib import Path

import httpx
import pytest

_TEST_DIR = Path(tempfile.mkdtemp(prefix="hunjing_pytest_"))

# ── 数据库选择 ────────────────────────────────────────────────
# 如果环境变量 HUNJING_TEST_DATABASE_URL 已设，走 PostgreSQL；
# 否则使用 SQLite 临时文件。
_HUNJING_TEST_DATABASE_URL = os.environ.get("HUNJING_TEST_DATABASE_URL", "")

if _HUNJING_TEST_DATABASE_URL:
    os.environ["DATABASE_URL"] = _HUNJING_TEST_DATABASE_URL
    _TEST_DB_PATH = None
else:
    _TEST_DB_PATH = _TEST_DIR / "test_hunjing.sqlite3"
    os.environ["SQLITE_DATABASE_PATH"] = str(_TEST_DB_PATH)

os.environ["JWT_SECRET"] = "test-jwt-secret-for-testing-pytest"

# 预热模型注册（必须在上述环境变量设置之后导入，避免 settings 读到开发库配置）：
# 先完整初始化 app.models，避免测试模块直接导入域模型时
# 触发 app.db.base ↔ app.models 的部分初始化循环
import app.models  # noqa: E402,F401


def _assert_test_database_url(database_url: str) -> None:
    if database_url.startswith("sqlite+aiosqlite:///"):
        if "test" not in database_url and "pytest" not in database_url:
            raise AssertionError("测试 SQLite 数据库 URL 必须指向测试路径。")
    elif database_url.startswith("postgresql"):
        if "test" not in database_url.lower():
            raise AssertionError("测试 PostgreSQL 数据库 URL 必须指向测试数据库（名称含 test）。")
    else:
        raise AssertionError(f"不支持的测试数据库类型：{database_url}")


@pytest.fixture(autouse=True)
async def reset_database() -> AsyncGenerator[None, None]:
    from app.db.base import Base
    from app.db.init_data import seed_default_data
    from app.db.session import AsyncSessionLocal, configure_database
    from app.db.session import engine as _engine

    _assert_test_database_url(str(_engine.url))
    # PG 模式下重新初始化 engine，避免连接池在旧的 event loop 上下文中
    if str(_engine.url).startswith("postgresql"):
        from app.core.config import settings

        configure_database(settings.resolved_database_url)
        from app.db.session import engine as _engine

    async with _engine.begin() as conn:
        if str(_engine.url).startswith("postgresql"):
            # PG: 显式 CASCADE + 断开连接池，避免 FK 约束阻塞
            await conn.run_sync(Base.metadata.drop_all)
            await _engine.dispose()
            configure_database(str(_engine.url))
            from app.db.session import engine as _engine
        else:
            await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncSessionLocal() as session:
        await seed_default_data(session)
    yield
    _assert_test_database_url(str(_engine.url))
    async with _engine.begin() as conn:
        if str(_engine.url).startswith("postgresql"):
            await conn.run_sync(Base.metadata.drop_all)
            await _engine.dispose()
        else:
            await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def auth_headers() -> dict[str, str]:
    """创建测试用户并返回 ``Authorization: Bearer <token>`` 请求头。"""
    import uuid

    from app.core.ids import new_user_id
    from app.db.session import AsyncSessionLocal
    from app.models.user import User
    from app.services.auth import create_access_token, hash_password
    from app.services.characters import ensure_default_character_for_owner

    async with AsyncSessionLocal() as session:
        suffix = uuid.uuid4().hex[:8]
        user = User(
            id=new_user_id(),
            username=f"tester_{suffix}",
            email=f"{suffix}@test.local",
            password_hash=hash_password("pass"),
            is_active=True,
            is_admin=False,
        )
        session.add(user)
        await session.flush()
        user_id = user.id
        await ensure_default_character_for_owner(session, user_id)
        await session.commit()
    token = create_access_token(subject=user_id, extra={"owner_scope": user_id})
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
async def auth_client(auth_headers: dict[str, str]) -> AsyncIterator[httpx.AsyncClient]:
    """带 JWT 认证的 HTTP 客户端（默认 Header 注入 Bearer token）。"""
    from httpx import ASGITransport, AsyncClient

    from app.main import app

    transport = ASGITransport(app=app)
    async with AsyncClient(
        transport=transport,
        base_url="http://test",
        headers=auth_headers,
    ) as http_client:
        yield http_client


@pytest.fixture
async def admin_headers() -> dict[str, str]:
    """创建管理员测试用户并返回认证请求头。"""
    import uuid

    from app.core.ids import new_user_id
    from app.db.session import AsyncSessionLocal
    from app.models.user import User
    from app.services.auth import create_access_token, hash_password
    from app.services.characters import ensure_default_character_for_owner

    async with AsyncSessionLocal() as session:
        suffix = uuid.uuid4().hex[:8]
        user = User(
            id=new_user_id(),
            username=f"admin_{suffix}",
            email=f"admin_{suffix}@test.local",
            password_hash=hash_password("pass"),
            is_active=True,
            is_admin=True,
        )
        session.add(user)
        await session.flush()
        user_id = user.id
        await ensure_default_character_for_owner(session, user_id)
        await session.commit()
    token = create_access_token(subject=user_id, extra={"owner_scope": user_id})
    return {"Authorization": f"Bearer {token}"}