mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 05:21:57 +00:00
ThearchyHelios
2e7d6103ea
- 配置：database_url、checkpoint_url（启动时校验写法，只认 PostgreSQL）、连接池四项；启动日志写用的是哪种库（主机、库名、用户，不写密码） - 引擎（db/base.make_engine）：SQLite 照旧 WAL、外键、忙等；PostgreSQL 用连接池、探活、会话时区 UTC，装上方言调整（db/postgres.py）：建表时 String(n) 是不限长的 VARCHAR（SQLite 从不查长度，导过来的存量也可能超长）、ORDER BY 里可空列的 NULL 位置照 SQLite 补写（升序在前、降序在后，窗口函数里同样）；Integer 列在 PostgreSQL 上是 BIGINT（列挂到表上时换成 with_variant，绑定参数跟着列按 BIGINT 转换）；JSON 列两边都存原文（PostgreSQL 的 json，不用 JSONB：JSONB 重排对象里的键，快照的列序、表的先后读回来就变了）；JSON 里的 NaN 写成 null、NUL 去掉（PostgreSQL 不收） - 迁移按方言：已有的列和索引用 inspector 认；PostgreSQL 上补列的类型取模型里这一列的、缺省值取迁移表里的（布尔 0/1 换成 false/true），ADD COLUMN IF NOT EXISTS；去重语句里的布尔按方言写；部分唯一索引补 postgresql_where；init_db 在 PostgreSQL 上拿咨询锁（几个进程同时启动不撞） - 应用库上的 SQLite 专用写法改成按方言编译的表达式（db/portable.py）：json_extract 的十几处（口径召回、出具、对账、个人应用、幂等键、访问控制、数据源变化通知、剖析任务、待办超时）换成 json_eq / json_flag / json_text，json_type 换成 json_is_string（两种库同一套带类型的比较）；能力登记的 sqlite insert 换成按会话方言的 insert_ignore；能力使用次数的 json_each 按方言分支；附件「没绑记录」的条件不靠 json 的等号；按报错文字认唯一约束的地方换成 unique_violation（按表和列认，两种库的报错都认得） - 断点（engine/checkpoints.py）：PostgreSQL 时开 psycopg 连接池、在咨询锁里 setup；runner 只认 BaseCheckpointSaver，读断点仍只走 saver 的公开接口 - 依赖：新 extra postgres（asyncpg、psycopg[binary] 3、psycopg-pool、langgraph-checkpoint-postgres 3.1.2）；langgraph-checkpoint 卡在 <4.3（断点格式就在这个包里，postgres 包新版要 4.3，不动已装的 4.2.0）；锁文件补上新包 - 运行收尾：记 action.unknown、安排核实做完才把运行摘掉（以前先摘掉，等「不在跑了」再读事件的一方在库慢时读不到这一条） - 测试：AGENTLAB_TEST_DATABASE_URL 时跑在 PostgreSQL 上（每个 xdist 进程建自己的库、按事件循环分的连接池 tests/pgpool.py），不设时盖掉环境里的数据库地址照旧用 SQLite；scripts/test-postgres.sh 起临时的 postgres:16 容器跑全量；sqlite_only 标记；另开连接改库的几个竞态测试、触发器测试改成两种库都能跑（tests/appdb.py）；新增 test_db_portable 守着方言调整、JSON 键序、超长字符串、大整数、NULL 排序、时区、PostgreSQL 老库迁移
114 lines
4.7 KiB
Python
114 lines
4.7 KiB
Python
"""把整个测试会话关进一个临时目录。
|
||
|
||
不这么做的话，任何一个直接用 SessionLocal 的测试都会往开发库
|
||
（data/agentlab.db）里写行——写进去的 run 会出现在运行列表里，和真实运行
|
||
混在一起，而且带着测试的假数据。这不是假想：test_emit_order 第一版就这么
|
||
干了，几行 emit-order-* 的假运行直接排在了列表最前面。
|
||
|
||
必须在 app.* 被 import 之前设好环境变量：settings 是模块级单例，engine 在
|
||
app.db.base 导入时就按 db_url 建好了。conftest 早于测试模块加载，所以这里
|
||
是唯一还来得及的地方。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import atexit
|
||
import os
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
_TMP = Path(tempfile.mkdtemp(prefix="agentlab-tests-"))
|
||
os.environ["AGENTLAB_DATA_DIR"] = str(_TMP)
|
||
|
||
|
||
def _postgres_database(admin_url: str) -> str:
|
||
"""AGENTLAB_TEST_DATABASE_URL 给的是管理用的库：每个 xdist 进程在同一台服务器上建一个自己的空库（库名加进程号后缀，
|
||
已有的同名库先删掉），进程退出时删掉。返回这个库的地址。scripts/test-postgres.sh 起临时容器、设好这个变量。"""
|
||
import psycopg
|
||
from sqlalchemy.engine import make_url
|
||
|
||
from app.db.urls import libpq_conninfo, normalize_database_url
|
||
|
||
url = make_url(normalize_database_url(admin_url))
|
||
name = f"{url.database}_{os.environ.get('PYTEST_XDIST_WORKER', 'main')}"
|
||
admin = libpq_conninfo(url.render_as_string(hide_password=False))
|
||
|
||
def run(sql: str) -> None:
|
||
with psycopg.connect(admin, autocommit=True) as conn:
|
||
conn.execute(sql)
|
||
|
||
run(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')
|
||
run(f'CREATE DATABASE "{name}"')
|
||
|
||
def drop() -> None:
|
||
try:
|
||
run(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')
|
||
except Exception: # noqa: BLE001 - 服务器先走了（容器已删）就算了
|
||
pass
|
||
|
||
atexit.register(drop)
|
||
return url.set(database=name).render_as_string(hide_password=False)
|
||
|
||
|
||
# 应用库：缺省数据目录里的 SQLite。设了 AGENTLAB_TEST_DATABASE_URL 就跑在 PostgreSQL 上。不管哪种，都盖掉环境和 .env
|
||
# 里可能有的 AGENTLAB_DATABASE_URL / AGENTLAB_CHECKPOINT_URL：测试绝不能连到真在用的库上
|
||
_TEST_PG = os.environ.get("AGENTLAB_TEST_DATABASE_URL", "").strip()
|
||
os.environ["AGENTLAB_CHECKPOINT_URL"] = ""
|
||
os.environ["AGENTLAB_DATABASE_URL"] = _postgres_database(_TEST_PG) if _TEST_PG else ""
|
||
if _TEST_PG:
|
||
# 每个测试一个事件循环（pytest-asyncio 的缺省），asyncpg 的连接不能跨循环用：引擎先按不用连接池建，
|
||
# 再换成按事件循环分的池子（tests/pgpool.py），一个测试里的查库照样复用连接
|
||
os.environ["AGENTLAB_DB_POOL_SIZE"] = "0"
|
||
|
||
from app.db.base import engine as _app_engine # noqa: E402
|
||
from tests.pgpool import install as _loop_local_pool # noqa: E402
|
||
|
||
_loop_local_pool(_app_engine)
|
||
# 别把真实的 key 带进测试环境
|
||
os.environ.setdefault("AGENTLAB_SECRET_KEY", "test-only-not-a-real-key")
|
||
# 定时任务的调度循环不在测试里起：走 lifespan 的测试不该到点替人发起运行。要测调度的直接调它
|
||
os.environ.setdefault("AGENTLAB_SCHEDULER", "0")
|
||
|
||
import pytest # noqa: E402
|
||
|
||
|
||
@pytest.fixture(scope="session", autouse=True)
|
||
def _prepare_schema():
|
||
"""建表。测试库是全新的临时文件，什么都没有。"""
|
||
import asyncio
|
||
|
||
from app.db.base import Base, engine
|
||
|
||
async def create() -> None:
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(create())
|
||
yield
|
||
|
||
|
||
def pytest_report_header(config) -> str:
|
||
where = "PostgreSQL（AGENTLAB_TEST_DATABASE_URL）" if _TEST_PG else "SQLite"
|
||
return f"测试数据目录：{_TMP}；应用库：{where}"
|
||
|
||
|
||
def pytest_collection_modifyitems(config, items) -> None:
|
||
"""标了 sqlite_only 的测试只在应用库是 SQLite 时跑：它们断言的是 SQLite 自己的东西（数据目录里的库文件、PRAGMA）。"""
|
||
if not _TEST_PG:
|
||
return
|
||
skip = pytest.mark.skip(reason="只对 SQLite 应用库成立")
|
||
for item in items:
|
||
if "sqlite_only" in item.keywords:
|
||
item.add_marker(skip)
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def scenic_db(tmp_path_factory) -> str:
|
||
"""合成的景区业务库（tests/fixtures/catalog/scenic.py）：50 张表、真外键与只能靠命名推断的关系并存。
|
||
|
||
每个测试进程建一次，只读使用；要改库内容的测试自己调 scenic.build(tmp_path / "x.db") 另建一个。
|
||
"""
|
||
from tests.fixtures.catalog import scenic
|
||
|
||
return str(scenic.build(tmp_path_factory.mktemp("catalog") / "scenic.db"))
|