mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 20:51:59 +00:00
ThearchyHelios
7febf2bcdb
- 编排者的决定：PostgreSQL 下文本列写入时去掉 NUL 字符（before_cursor_execute 钩子，只装在应用库引擎上，看所有字符串参数，没有 NUL 的原样交回），SQLite 不变；同 JSON 列的处理 - R-a4 应用库地址里的 sslmode= 换成 asyncpg 的 ssl=（取值同一套），sslrootcert 等证书参数、application_name 这类 asyncpg 不收的参数启动时就报错并说怎么写（证书用 PGSSLROOTCERT 等环境变量）；断点库连接串换回 sslmode= - R-a6 不分大小写的搜索换成 db/portable.contains_ci：SQLite 上同 ilike（lower LIKE lower），PostgreSQL 上模式带 COLLATE "default"（列是 COLLATE "C"，列上的 ILIKE 只认 ASCII，É/é、Ж/ж、全角Ａ/ａ 匹配不上）；运行列表按名字、对象索引按注释两处都改；改正 db/postgres 里那句说明 - R-a7 JSON 守卫改成编译层：tests/jsonguard 挂在应用库编译器上（conftest，两种库、跑全量时一直生效），整个 JSON 值出现在 = / != / IN / IS DISTINCT FROM / 大小比较、SELECT DISTINCT、GROUP BY、ORDER BY、UNION（去重的几种）里就报错；按类型认，变量别名、跨行 order_by、select(Run.origin).distinct()、JSON 下标的比较都抓得到，测试拿这些反例证明，json_eq、IS NULL、UNION ALL 不受影响。原来的逐行正则测试删掉 - R-a10 json_eq 两种库同一语义、按 JSON 类型比：字符串只等于字符串，数只等于数（1 等于 1.0），True / False 只等于 JSON 的 true / false（SQLite 上用 json_type 分开，以前 1 能匹配 true）；非标量 TypeError。json_text 两种库都是「字符串取本身、别的取 JSON 写法」（SQLite 上同 PostgreSQL 的 ->>），数 1 的 json_text 两边都是 '1' - db/postgres 说明里写清 NULL 排序补写的范围（只认直接按列排的）
120 lines
5.1 KiB
Python
120 lines
5.1 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
|
||
|
||
# 整个 JSON 值做比较、去重、分组、排序时编译就报错（tests/jsonguard.py）：PostgreSQL 的 json 不支持，SQLite 的测试也拦下
|
||
from app.db.base import engine as _guarded_engine # noqa: E402
|
||
from tests.jsonguard import install as _json_guard # noqa: E402
|
||
|
||
_json_guard(_guarded_engine.sync_engine.dialect)
|
||
|
||
|
||
@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"))
|