133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
|
"""在独立测试库上跑测试：与开发用的 worker 进程物理隔离。
|
|||
|
|
|
|||
|
|
为什么需要：`python -m app.worker` 会认领任意 `queued/running/cancel_requested` 运行，
|
|||
|
|
并消费 Outbox 事件。只要它与测试共用同一个库，测试造的数据就会被并发消费，表现为
|
|||
|
|
"同一个用例全量跑失败、单跑通过"的偶发失败——那不是代码缺陷，是环境干扰。
|
|||
|
|
|
|||
|
|
流程（每一步都只作用于测试库）：
|
|||
|
|
|
|||
|
|
1. 由 `.env` 的 `MYSQL_DSN` 推导测试库 DSN（只换库名，其余原样）；
|
|||
|
|
2. 测试库为空时先导入 `alembic/baseline_generated.sql`（33 张历史基线表不在迁移链里，
|
|||
|
|
迁移链只负责基线之后的结构变更）；
|
|||
|
|
3. 对测试库执行 `alembic upgrade head`；
|
|||
|
|
4. 灌入 RBAC 测试账号（`tools/seed_test_rbac.py`，9001/9002/9003 号段）；
|
|||
|
|
5. 仅在本进程内用 `MYSQL_DSN` 覆盖，跑 pytest。
|
|||
|
|
|
|||
|
|
前置（只需一次，用有建库权限的账号执行）：
|
|||
|
|
|
|||
|
|
CREATE DATABASE IF NOT EXISTS jr_agent_test
|
|||
|
|
CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
|
|||
|
|
GRANT ALL PRIVILEGES ON jr_agent_test.* TO 'jr_app'@'localhost';
|
|||
|
|
|
|||
|
|
用法：
|
|||
|
|
|
|||
|
|
python -m tools.run_tests_on_test_db # 默认跑 tests/integration
|
|||
|
|
python -m tools.run_tests_on_test_db tests # 跑全部测试目录
|
|||
|
|
python -m tools.run_tests_on_test_db tests/unit --ignore=... # 透传 pytest 参数
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import subprocess
|
|||
|
|
import sys
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pymysql
|
|||
|
|
from sqlalchemy.engine import make_url
|
|||
|
|
|
|||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|||
|
|
if str(ROOT) not in sys.path:
|
|||
|
|
sys.path.insert(0, str(ROOT))
|
|||
|
|
|
|||
|
|
from app.core.config import get_settings # noqa: E402
|
|||
|
|
|
|||
|
|
BASELINE_SQL = ROOT / "alembic" / "baseline_generated.sql"
|
|||
|
|
DEFAULT_DATABASE = "jr_agent_test"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _connect(**overrides: object) -> pymysql.connections.Connection:
|
|||
|
|
settings = get_settings()
|
|||
|
|
url = make_url(settings.mysql_dsn)
|
|||
|
|
return pymysql.connect(
|
|||
|
|
host=url.host or "127.0.0.1",
|
|||
|
|
port=url.port or 3306,
|
|||
|
|
user=url.username,
|
|||
|
|
password=url.password or "",
|
|||
|
|
charset="utf8mb4",
|
|||
|
|
**overrides, # type: ignore[arg-type]
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _table_count(connection: pymysql.connections.Connection, database: str) -> int:
|
|||
|
|
with connection.cursor() as cursor:
|
|||
|
|
cursor.execute(
|
|||
|
|
"SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s",
|
|||
|
|
(database,),
|
|||
|
|
)
|
|||
|
|
return int(cursor.fetchone()[0])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _import_baseline(connection: pymysql.connections.Connection, database: str) -> None:
|
|||
|
|
"""逐条执行基线 SQL。
|
|||
|
|
|
|||
|
|
不用 `CLIENT.MULTI_STATEMENTS` 一次发整脚本：实测那条路径下语句没有真正生效
|
|||
|
|
（连接正常返回、表却没建出来），逐条执行才能看到真实结果与失败位置。
|
|||
|
|
"""
|
|||
|
|
script = BASELINE_SQL.read_text(encoding="utf-8")
|
|||
|
|
statements = [item.strip() for item in script.split(";") if item.strip()]
|
|||
|
|
with connection.cursor() as cursor:
|
|||
|
|
cursor.execute(f"USE `{database}`")
|
|||
|
|
for statement in statements:
|
|||
|
|
cursor.execute(statement)
|
|||
|
|
connection.commit()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _run(step: str, args: list[str], env: dict[str, str]) -> None:
|
|||
|
|
print(f"\n>>> {step}: {' '.join(args)}", flush=True)
|
|||
|
|
completed = subprocess.run(args, cwd=ROOT, env=env, check=False)
|
|||
|
|
if completed.returncode != 0:
|
|||
|
|
raise SystemExit(f"{step} 失败（退出码 {completed.returncode}）")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> None:
|
|||
|
|
database = os.environ.get("TEST_DB_NAME", DEFAULT_DATABASE)
|
|||
|
|
pytest_args = sys.argv[1:] or ["tests/integration"]
|
|||
|
|
|
|||
|
|
base_url = make_url(get_settings().mysql_dsn)
|
|||
|
|
test_url = base_url.set(database=database)
|
|||
|
|
if base_url.database == database:
|
|||
|
|
raise SystemExit(f"测试库不能与开发库同名（当前：{database}）")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
connection = _connect(database=database)
|
|||
|
|
except pymysql.err.OperationalError as exc:
|
|||
|
|
raise SystemExit(
|
|||
|
|
f"无法连接测试库 {database}：{exc}\n"
|
|||
|
|
"请先用有建库权限的账号执行：\n"
|
|||
|
|
f" CREATE DATABASE IF NOT EXISTS {database} "
|
|||
|
|
"CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;\n"
|
|||
|
|
f" GRANT ALL PRIVILEGES ON {database}.* TO 'jr_app'@'localhost';"
|
|||
|
|
) from exc
|
|||
|
|
|
|||
|
|
with connection:
|
|||
|
|
tables = _table_count(connection, database)
|
|||
|
|
print(f"测试库 {database} 现有表：{tables}")
|
|||
|
|
if tables == 0:
|
|||
|
|
print("测试库为空，先导入历史基线表（alembic/baseline_generated.sql）")
|
|||
|
|
_import_baseline(connection, database)
|
|||
|
|
print(f"基线导入完成，现有表：{_table_count(connection, database)}")
|
|||
|
|
|
|||
|
|
env = {
|
|||
|
|
**os.environ,
|
|||
|
|
"MYSQL_DSN": test_url.render_as_string(hide_password=False),
|
|||
|
|
}
|
|||
|
|
python = sys.executable
|
|||
|
|
_run("结构迁移", [python, "-m", "alembic", "upgrade", "head"], env)
|
|||
|
|
_run("灌入测试账号", [python, "-m", "tools.seed_test_rbac"], env)
|
|||
|
|
_run("执行测试", [python, "-m", "pytest", *pytest_args, "-q"], env)
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|