126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
"""pytest 数据库 fixture：会话只准备迁移模板，每例仍独占数据库。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from .数据库底座 import PostgreSQL后端
|
||
from .数据库底座 import 数据库底座 as 底座管理器
|
||
|
||
_会话证据 = pytest.StashKey[tuple[底座管理器, Path]]()
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def 隔离数据库URL():
|
||
值 = os.environ.get("MUSE_TEST_DATABASE_URL", "").strip()
|
||
if not 值:
|
||
pytest.fail("未设置 MUSE_TEST_DATABASE_URL；隔离 PG 验证未执行", pytrace=False)
|
||
return 值
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def 数据库底座(隔离数据库URL, tmp_path_factory, request):
|
||
if not request.config.getoption("--外部环境"):
|
||
pytest.fail("数据库夹具必须显式选择 --外部环境", pytrace=False)
|
||
目录 = tmp_path_factory.mktemp("数据库底座")
|
||
根 = Path(__file__).resolve().parents[2]
|
||
基础目录 = tmp_path_factory.getbasetemp()
|
||
if os.environ.get("PYTEST_XDIST_WORKER"):
|
||
基础目录 = 基础目录.parent
|
||
共享目录 = 基础目录 / "数据库角色初始化"
|
||
底座 = 底座管理器(
|
||
PostgreSQL后端(隔离数据库URL, 共享目录),
|
||
根,
|
||
目录,
|
||
worker=os.environ.get("PYTEST_XDIST_WORKER", "main"),
|
||
)
|
||
输出 = 目录 / "成本事件.json"
|
||
request.config.stash[_会话证据] = (底座, 输出)
|
||
try:
|
||
yield 底座
|
||
finally:
|
||
try:
|
||
底座.关闭()
|
||
finally:
|
||
输出.write_text(json.dumps(底座.事件, ensure_ascii=False, indent=2))
|
||
终端 = request.config.pluginmanager.getplugin("terminalreporter")
|
||
if 终端:
|
||
终端.write_line(f"数据库成本事件：{输出}")
|
||
|
||
|
||
@pytest.fixture
|
||
def 测试数据库端点(数据库底座, tmp_path):
|
||
"""供真实初始化目标使用；这里不创建没有用途的应用库。"""
|
||
return 数据库底座.端点(tmp_path / "数据库端点")
|
||
|
||
|
||
@pytest.fixture
|
||
def 应用测试库(数据库底座, tmp_path):
|
||
借用 = 数据库底座.借库 if os.environ.get("MUSE_ISOLATED_CLONE") == "1" else 数据库底座.共享库
|
||
with 借用(tmp_path / "连接引用") as 工厂:
|
||
yield 工厂
|
||
|
||
|
||
@pytest.fixture
|
||
def 空测试库(数据库底座, tmp_path):
|
||
with 数据库底座.借库(tmp_path / "空库引用", 空库=True) as 工厂:
|
||
yield 工厂
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def 内置种子方案():
|
||
# 延迟加载，纯离线用例不导入资源，更不建立数据库连接。
|
||
from muse.元数据.接口 import 元数据服务, 导入内置结构
|
||
from muse.共享.调用身份 import 用途
|
||
|
||
根 = Path(__file__).resolve().parents[2]
|
||
# 种子及导入器是基线身份的一部分，禁止模板跨不同种子逻辑复用。
|
||
路径集 = sorted((根 / "src/muse/元数据/内置结构").glob("*.yaml"))
|
||
if not 路径集:
|
||
raise RuntimeError("内置结构种子目录为空")
|
||
摘要 = hashlib.sha256()
|
||
for 路径 in [*路径集, 根 / "src/muse/元数据/内置导入.py"]:
|
||
摘要.update(路径.relative_to(根).as_posix().encode())
|
||
摘要.update(路径.read_bytes())
|
||
|
||
def 导入(工厂):
|
||
with 工厂[用途.维护].连接() as 连, 连.transaction():
|
||
导入内置结构(元数据服务(连))
|
||
|
||
return 摘要.hexdigest(), 导入
|
||
|
||
|
||
@pytest.fixture
|
||
def 内置结构测试库(数据库底座, 内置种子方案, tmp_path):
|
||
指纹, 导入 = 内置种子方案
|
||
借用 = 数据库底座.借库 if os.environ.get("MUSE_ISOLATED_CLONE") == "1" else 数据库底座.共享库
|
||
with 借用(tmp_path / "种子库引用", 种子指纹=指纹, 导入种子=导入) as 工厂:
|
||
yield 工厂
|
||
|
||
|
||
@pytest.hookimpl(hookwrapper=True)
|
||
def pytest_runtest_makereport(item, call):
|
||
结果 = yield
|
||
报告 = 结果.get_result()
|
||
底座 = item.funcargs.get("数据库底座")
|
||
if 底座 is not None:
|
||
底座.记录(
|
||
"phase",
|
||
nodeid=item.nodeid,
|
||
phase=报告.when,
|
||
seconds=报告.duration,
|
||
outcome=报告.outcome,
|
||
)
|
||
|
||
|
||
def pytest_sessionfinish(session, exitstatus):
|
||
# 最后一例的 teardown 报告晚于 session fixture 清理；在此补齐阶段耗时。
|
||
if _会话证据 in session.config.stash:
|
||
底座, 输出 = session.config.stash[_会话证据]
|
||
输出.write_text(json.dumps(底座.事件, ensure_ascii=False, indent=2))
|