vent_agent_2.0/tests/conftest.py

"""pytest 全局装置。

**必须在导入任何项目模块之前设置环境变量**：``config`` 在导入时就读
``.env`` 与环境变量并定稿，``db.db_conn`` 在导入时就建好连接适配器。
所以下面的 ``os.environ`` 赋值必须在 ``import config`` 之前发生——
这也是把它们放在文件最顶部、并设 ``DEEP_AGENT_SKIP_DOTENV`` 的原因：
不跳过 .env 的话，开发者本机的 .env 会覆盖这里的测试配置，
同一套测试在不同机器上跑出不同结果。
"""

from __future__ import annotations

import os
import shutil
import tempfile
import uuid
from pathlib import Path

# ---------- 测试环境（必须在项目导入前设置）----------
# 用 setdefault 而非直接赋值：外部已给出的 DB_DIALECT / DB_* 优先生效，
# 这样同一套测试可以直接跑在 MySQL 上验证方言无关性——
#   DB_DIALECT=mysql DB_HOST=127.0.0.1 DB_PORT=3399 DB_USER=root DB_NAME=t python -m pytest
_TMP_ROOT = Path(tempfile.mkdtemp(prefix="deep_agent_test_"))
os.environ["DEEP_AGENT_ENV_PRIORITY"] = "none"   # 完全忽略本机 .env
os.environ.setdefault("DB_DIALECT", "sqlite")
os.environ.setdefault("DB_SQLITE_PATH", str(_TMP_ROOT / "test.sqlite"))
os.environ.setdefault("CHECKPOINTER", "memory")   # 测试不需要落盘图状态
os.environ["CTX_COMPACT_ENABLED"] = "0"           # 不引入摘要中间件，测试更纯粹
os.environ["OPENAI_API_KEY"] = "sk-test-key"
os.environ["OPENAI_BASE_URL"] = "https://example.invalid"
os.environ["TFAGENTS_DAILY_TOKEN_LIMIT"] = "0"
os.environ["BILLING_ALLOW_UNPRICED"] = "0"
os.environ["LOG_LEVEL"] = "WARNING"

import pytest  # noqa: E402

import config  # noqa: E402
from db import db_conn  # noqa: E402


@pytest.fixture(scope="session", autouse=True)
def _schema() -> None:
    """整个测试会话共用一份建好的表结构。"""
    db_conn.init_schema()


@pytest.fixture(autouse=True)
def _clean_tables():
    """每个用例前清空业务表，避免用例间互相污染。

    删表数据而不是重建 schema：重建的代价在 sqlite 上也不小，而这里
    只需要一个干净的状态。
    """
    for table in (
        "messages",
        "sessions",
        "session_context_usage",
        "context_archives",
        "billing_usage",
        "billing_pricing",
        "daily_quota",
        "user_preferences",
        "schedule_runs",
        "schedules",
    ):
        db_conn.execute(f"DELETE FROM {table}", ())
    # 工具注册表是模块级单例，清空后重新装载，避免用例顺序影响工具清单
    from tools import registry

    registry.reset_for_tests()
    registry.load_builtin_tools()
    yield


@pytest.fixture(scope="session")
def dialect() -> str:
    """当前测试跑的方言。用于需要区分方言的断言。"""
    return config.DB_DIALECT


@pytest.fixture
def pricing_ready():
    """确保单价表里有当前模型的条目，供计费相关用例使用。"""
    from billing import pricing

    pricing.seed_default_pricing()
    return pricing


@pytest.fixture
def session_id() -> str:
    """建一个测试会话，返回 session_id。"""
    from db import chat_store

    return chat_store.create_session("tester", "full", "测试会话")["id"]


@pytest.fixture
def unique_user() -> str:
    """每次调用返回一个新用户名，避免额度/流水跨用例串味。"""
    return f"u_{uuid.uuid4().hex[:8]}"


def pytest_sessionfinish(session, exitstatus) -> None:
    """收尾清理临时目录。"""
    shutil.rmtree(_TMP_ROOT, ignore_errors=True)