9e6f1c4481
- src/muse 新版全模块（装配/共享/上下文/任务运行/作品规划/故事世界/正文写作/审校修订/知识方法/作者经验/效果评测/交付连载/资料研究/正式变更/元数据/接入/基础设施/编排）+ 测试树（单元/契约/集成/架构/迁移/端到端/夹具） - 129 项功能全部实现与自动验证（功能覆盖.json/矩阵），含 W31 补齐的规则与代价/节奏安排/伏笔与承诺 - 旧实现按处置清单退出（702 条中 324 删，保护合同与未迁移条目留存有据）；web/app.py 旧工作台退役，新工作台为唯一写入口 - 数据库/旧库迁移：真实旧库内容批次迁移链（端点守卫/PG作品正文映射/质量资产缺省投影） - 运行手册 docs/运行手册.md；W30 本机服务阶段一已运行（infra PG 为正式内容权威） - R2 执行证据与私有运行材料在 .agents.local/改造/R2-20260909/（不入库）
709 lines
26 KiB
Python
709 lines
26 KiB
Python
"""原生 pytest 的用例身份、进程隔离与共享夹具。
|
||
|
||
默认使用 OS 防护覆盖收集、驱动及子进程。外部环境必须显式选择；
|
||
完整 ID 绑定登记文件及符号，参数行和 Junit 保留可追溯身份。
|
||
本文件自包含，临时项目复制后验证相同测试入口。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import ctypes.util
|
||
import errno
|
||
import json
|
||
import os
|
||
import socket
|
||
import sys
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 标记与用例选择
|
||
# ---------------------------------------------------------------------------
|
||
|
||
外部标记 = ("数据库", "网络", "真实模型", "浏览器", "宿主")
|
||
|
||
# 执行计数：供会话结束守卫判断“无实际执行”
|
||
_执行计数 = {"passed": 0, "skipped": 0, "failed": 0}
|
||
|
||
|
||
def pytest_addoption(parser: pytest.Parser) -> None:
|
||
组 = parser.getgroup("muse", "Muse 用例选择")
|
||
组.addoption("--用例清单", default=None, help="显式的完整用例 ID 与目标符号清单")
|
||
组.addoption(
|
||
"--外部环境", action="store_true", help="显式允许外部用例；仍按标记选择并使用隔离配置"
|
||
)
|
||
组.addoption(
|
||
"--case",
|
||
action="append",
|
||
default=None,
|
||
metavar="CASE_ID",
|
||
help="按稳定用例 ID（TC-/NC- 前缀）选择；可多次给出，语义为并集",
|
||
)
|
||
|
||
|
||
def pytest_configure(config: pytest.Config) -> None:
|
||
for 名 in 外部标记:
|
||
config.addinivalue_line("markers", f"{名}: 外部环境用例；见 tests/conftest.py")
|
||
|
||
|
||
def _读取用例清单(config: pytest.Config) -> list[dict[str, Any]]:
|
||
指定 = config.getoption("--用例清单")
|
||
路径 = (
|
||
Path(指定)
|
||
if 指定
|
||
else config.rootpath / "docs/系统架构/新版设计/验证设计/测试用例清单.json"
|
||
)
|
||
if not 路径.exists() and not 指定 and not config.getoption("--case"):
|
||
return []
|
||
try:
|
||
清单 = json.loads(路径.read_text(encoding="utf-8"))
|
||
return [目标 for 旧例 in 清单["cases"] for 目标 in 旧例.get("target_cases", [])] + 清单[
|
||
"new_cases"
|
||
]
|
||
except (OSError, ValueError, TypeError, KeyError) as exc:
|
||
raise pytest.UsageError(f"用例清单无效：{路径.name}（{type(exc).__name__}）") from exc
|
||
|
||
|
||
def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
|
||
登记 = _读取用例清单(config)
|
||
选择 = config.getoption("--case") or []
|
||
按ID = {条["case_id"]: 条 for 条 in 登记}
|
||
请求 = {str(值): str(值).split("[", 1)[0] for 值 in 选择}
|
||
for 值, 身份 in 请求.items():
|
||
if 身份 not in 按ID:
|
||
raise pytest.UsageError(f"--case {值} 未登记，必须使用完整用例 ID")
|
||
按目标: dict[tuple[str, str], list[dict]] = {}
|
||
for 条 in 登记:
|
||
按目标.setdefault((条["file"], 条["symbol"]), []).append(条)
|
||
保留, 排除, 命中 = [], [], set()
|
||
for 项 in items:
|
||
文件 = 项.path.relative_to(config.rootpath).as_posix()
|
||
符号 = getattr(项, "originalname", None) or 项.name.split("[", 1)[0]
|
||
参数 = getattr(getattr(项, "callspec", None), "id", None)
|
||
对应 = [
|
||
条["case_id"]
|
||
for 条 in 按目标.get((文件, 符号), [])
|
||
if not 条.get("parameter_ids") or 参数 in 条["parameter_ids"]
|
||
]
|
||
if len(对应) > 1:
|
||
raise pytest.UsageError(f"用例身份重叠：{项.nodeid} 对应 {对应}")
|
||
身份 = 对应[0] if 对应 else None
|
||
if 身份:
|
||
项.user_properties.append(("case_id", 身份))
|
||
if 参数 is not None:
|
||
项.user_properties.append(("parameter_id", 参数))
|
||
该项命中 = {
|
||
值
|
||
for 值, id_ in 请求.items()
|
||
if id_ == 身份 and ("[" not in 值 or 值 == f"{身份}[{参数}]")
|
||
}
|
||
命中.update(该项命中)
|
||
if not 选择 or 该项命中:
|
||
保留.append(项)
|
||
else:
|
||
排除.append(项)
|
||
if set(请求) - 命中:
|
||
raise pytest.UsageError(f"--case 已登记但未收集：{sorted(set(请求) - 命中)}")
|
||
if 排除:
|
||
config.hook.pytest_deselected(items=排除)
|
||
items[:] = 保留
|
||
|
||
|
||
def pytest_sessionstart(session: pytest.Session) -> None:
|
||
"""整个离线进程使用 OS 防护，收集、libpq 与子进程共用相同边界。"""
|
||
if session.config.getoption("--外部环境"):
|
||
return
|
||
if sys.platform == "darwin":
|
||
库 = ctypes.CDLL("/usr/lib/libsandbox.dylib")
|
||
初始化 = 库.sandbox_init
|
||
初始化.argtypes = [ctypes.c_char_p, ctypes.c_uint64, ctypes.POINTER(ctypes.c_char_p)]
|
||
初始化.restype = ctypes.c_int
|
||
错误 = ctypes.c_char_p()
|
||
if 初始化(b"(version 1)(allow default)(deny network*)", 0, ctypes.byref(错误)):
|
||
raise pytest.UsageError("操作系统离线隔离初始化失败")
|
||
elif sys.platform.startswith("linux"):
|
||
路径 = ctypes.util.find_library("seccomp")
|
||
if not 路径:
|
||
raise pytest.UsageError("离线测试需要系统 libseccomp；不能无防护运行")
|
||
库 = ctypes.CDLL(路径)
|
||
库.seccomp_init.argtypes = [ctypes.c_uint32]
|
||
库.seccomp_init.restype = ctypes.c_void_p
|
||
库.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p]
|
||
库.seccomp_rule_add.argtypes = [
|
||
ctypes.c_void_p,
|
||
ctypes.c_uint32,
|
||
ctypes.c_int,
|
||
ctypes.c_uint,
|
||
]
|
||
库.seccomp_load.argtypes = [ctypes.c_void_p]
|
||
库.seccomp_release.argtypes = [ctypes.c_void_p]
|
||
上下文 = 库.seccomp_init(0x7FFF0000)
|
||
if not 上下文:
|
||
raise pytest.UsageError("无法创建离线系统调用规则")
|
||
try:
|
||
# 禁止建立网络 socket 和所有外部 connect；保留事件循环的内部 socketpair。
|
||
class 参数条件(ctypes.Structure):
|
||
_fields_ = [
|
||
("arg", ctypes.c_uint),
|
||
("op", ctypes.c_int),
|
||
("datum_a", ctypes.c_uint64),
|
||
("datum_b", ctypes.c_uint64),
|
||
]
|
||
|
||
库.seccomp_rule_add_array.argtypes = [
|
||
ctypes.c_void_p,
|
||
ctypes.c_uint32,
|
||
ctypes.c_int,
|
||
ctypes.c_uint,
|
||
ctypes.POINTER(参数条件),
|
||
]
|
||
拒绝 = 0x00050000 | errno.EPERM
|
||
connect = 库.seccomp_syscall_resolve_name(b"connect")
|
||
socket_call = 库.seccomp_syscall_resolve_name(b"socket")
|
||
if 库.seccomp_rule_add(上下文, 拒绝, connect, 0):
|
||
raise pytest.UsageError("无法登记离线连接规则")
|
||
for family in (socket.AF_INET, socket.AF_INET6):
|
||
条件 = 参数条件(0, 4, family, 0) # SCMP_CMP_EQ
|
||
if 库.seccomp_rule_add_array(上下文, 拒绝, socket_call, 1, ctypes.byref(条件)):
|
||
raise pytest.UsageError("无法登记离线网络规则")
|
||
if 库.seccomp_load(上下文):
|
||
raise pytest.UsageError("操作系统离线隔离初始化失败")
|
||
finally:
|
||
库.seccomp_release(上下文)
|
||
else:
|
||
raise pytest.UsageError("当前系统尚无离线进程隔离实现；使用受支持的 macOS 或 Linux 环境")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 离线防护：未显式选择外部环境的测试禁止一切 socket 连接
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class 离线连接被拒(RuntimeError):
|
||
"""离线测试尝试建立网络或数据库连接。"""
|
||
|
||
|
||
原始connect: Any = socket.socket.connect
|
||
原始connect_ex: Any = socket.socket.connect_ex
|
||
|
||
|
||
def _拒绝连接(self: socket.socket, address: Any) -> Any:
|
||
raise 离线连接被拒(
|
||
f"离线测试尝试连接 {address!r}；需要外部环境时显式标记 {'/'.join(外部标记)} 并使用对应夹具"
|
||
)
|
||
|
||
|
||
def _安装防护() -> None:
|
||
socket.socket.connect = _拒绝连接 # type: ignore[method-assign]
|
||
socket.socket.connect_ex = _拒绝连接 # type: ignore[method-assign]
|
||
|
||
|
||
def _卸载防护() -> None:
|
||
socket.socket.connect = 原始connect # type: ignore[method-assign]
|
||
socket.socket.connect_ex = 原始connect_ex # type: ignore[method-assign]
|
||
|
||
|
||
def _需要外部环境(项: pytest.Item) -> bool:
|
||
标记 = {m.name for m in 项.iter_markers()}
|
||
return any(名 in 标记 for 名 in 外部标记)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def 离线防护(request: pytest.FixtureRequest):
|
||
"""默认给所有未选择外部环境的用例安装 socket 防护；用例结束后恢复。"""
|
||
if _需要外部环境(request.node):
|
||
yield
|
||
return
|
||
_安装防护()
|
||
try:
|
||
yield
|
||
finally:
|
||
_卸载防护()
|
||
|
||
|
||
def pytest_runtest_logreport(report: pytest.TestReport) -> None:
|
||
if report.when == "call":
|
||
if report.passed:
|
||
_执行计数["passed"] += 1
|
||
elif report.failed:
|
||
_执行计数["failed"] += 1
|
||
elif report.skipped:
|
||
_执行计数["skipped"] += 1
|
||
elif report.when == "setup" and report.skipped:
|
||
_执行计数["skipped"] += 1
|
||
|
||
|
||
def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
|
||
"""全跳过不算通过：没有实际执行时改写退出码为 5（无测试运行）。"""
|
||
if exitstatus == 0 and _执行计数["passed"] == 0 and _执行计数["skipped"] > 0:
|
||
session.exitstatus = 5
|
||
print(
|
||
f"\n守卫：{_执行计数['skipped']} 个用例全部跳过、零实际执行；不计通过（退出码 5）",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 外部环境夹具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
数据库环境变量 = "MUSE_TEST_DATABASE_URL"
|
||
|
||
|
||
@pytest.fixture
|
||
def 隔离数据库URL() -> str:
|
||
"""隔离 PostgreSQL 连接串；缺失时跳过（跳过不计通过）。"""
|
||
值 = os.environ.get(数据库环境变量, "").strip()
|
||
if not 值:
|
||
pytest.skip(
|
||
f"未设置 {数据库环境变量}；隔离库用例未执行（不计通过）。启动本地库并导出该变量后重跑。"
|
||
)
|
||
return 值
|
||
|
||
|
||
@pytest.fixture
|
||
def 夹具根() -> Path:
|
||
"""tests/夹具 的绝对路径；供用例读取稳定回归输入。"""
|
||
根 = Path(__file__).resolve().parent / "夹具"
|
||
assert 根.is_dir(), f"夹具目录不存在：{根}"
|
||
return 根
|
||
|
||
|
||
@pytest.fixture
|
||
def 应用测试库(隔离数据库URL: str, tmp_path: Path):
|
||
"""跨模块接入测试使用全新库；返回受控文件引用，全部数据为合成数据。"""
|
||
import uuid
|
||
|
||
import psycopg
|
||
from psycopg import sql
|
||
|
||
from muse.共享.调用身份 import 用途
|
||
from muse.基础设施.数据库.迁移 import 执行迁移
|
||
from muse.基础设施.数据库.连接 import 数据库工厂
|
||
from muse.配置 import 数据库引用
|
||
|
||
根 = Path(__file__).resolve().parents[1]
|
||
库名 = "muse_http_" + uuid.uuid4().hex[:12]
|
||
with psycopg.connect(隔离数据库URL, autocommit=True) as 管理:
|
||
管理.execute((根 / "数据库/初始化/用途角色.sql").read_text())
|
||
管理.execute(sql.SQL("CREATE DATABASE {} OWNER muse_maint").format(sql.Identifier(库名)))
|
||
工厂 = {}
|
||
try:
|
||
for 名, 角色 in [
|
||
(用途.维护, "muse_maint"),
|
||
(用途.生产, "muse_app"),
|
||
(用途.评测, "muse_eval"),
|
||
]:
|
||
参数 = psycopg.conninfo.conninfo_to_dict(隔离数据库URL)
|
||
参数.update(dbname=库名, user=角色)
|
||
文件 = tmp_path / f"{名.value}.txt"
|
||
文件.write_text(psycopg.conninfo.make_conninfo(**参数))
|
||
文件.chmod(0o600)
|
||
工厂[名] = 数据库工厂(数据库引用("受控存储", str(文件)), 名)
|
||
with 工厂[用途.维护].连接() as 连:
|
||
执行迁移(连, 根 / "数据库/迁移")
|
||
yield 工厂
|
||
finally:
|
||
with psycopg.connect(隔离数据库URL, autocommit=True) as 管理:
|
||
管理.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(库名)))
|
||
|
||
|
||
# ---------------------------------------------------------------- W16 章后派生共享环境
|
||
|
||
|
||
class 合成派生计价:
|
||
版本 = "synthetic-price-1"
|
||
|
||
def 金额(self, 结果):
|
||
return Decimal("0.125") if 结果.用量 else None
|
||
|
||
|
||
class _离线派生配置检查:
|
||
身份 = "synthetic-derivation-validation"
|
||
|
||
def 验证(self, 内容, 执行用途):
|
||
from muse.任务运行.模型 import 内容哈希
|
||
from muse.任务运行.配置版本 import 配置验证证据
|
||
|
||
return 配置验证证据(
|
||
内容哈希(内容.冻结()),
|
||
内容.角色策略版本,
|
||
内容.资源发布身份,
|
||
执行用途,
|
||
("synthetic-derivation-contract",),
|
||
"runtime",
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def 章后环境(应用测试库, tmp_path):
|
||
"""作品、章后流程、抽取配置与合成宿主一次装配；抽取剧本由用例注入。"""
|
||
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from threading import Thread
|
||
|
||
from muse.任务运行.角色策略 import 角色策略目录
|
||
from muse.任务运行.配置版本 import (
|
||
凭据引用,
|
||
提供方配置,
|
||
运行配置内容,
|
||
配置版本管理,
|
||
)
|
||
from muse.任务运行.预算管理 import 预算管理, 额度策略
|
||
from muse.作品规划.接口 import 档案保存
|
||
from muse.元数据.接口 import 元数据服务, 导入内置结构
|
||
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
|
||
from muse.启动 import 构建, 装配章后处理
|
||
from muse.配置 import 应用配置
|
||
|
||
库 = 应用测试库[用途.生产]
|
||
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
|
||
导入内置结构(元数据服务(连))
|
||
策略 = 角色策略目录.从发布包()
|
||
装配 = 构建(
|
||
应用配置(
|
||
库.引用,
|
||
策略.资源发布身份,
|
||
运行用途=库.用途,
|
||
原文暂存=str(tmp_path / "raw"),
|
||
)
|
||
)
|
||
装配章后处理(装配, 合成派生计价(), 抽取配置ID="post-config")
|
||
|
||
收到: list[dict] = []
|
||
剧本: list = []
|
||
|
||
def 回应(数据):
|
||
收到.append(数据)
|
||
动作 = 剧本.pop(0) if 剧本 else {"类型": "空对象"}
|
||
if 动作.get("类型") == "失败":
|
||
raise RuntimeError("合成宿主按剧本失败")
|
||
输出 = [
|
||
{
|
||
"type": "message",
|
||
"content": [
|
||
{
|
||
"type": "output_text",
|
||
"text": json.dumps(动作.get("文本", {}), ensure_ascii=False),
|
||
}
|
||
],
|
||
}
|
||
]
|
||
载荷 = {
|
||
"type": "response.completed",
|
||
"response": {
|
||
"id": f"post-response-{len(收到)}",
|
||
"model": "claude-opus-4-8[1M]",
|
||
"status": "completed",
|
||
"output": 输出,
|
||
"usage": {"input_tokens": 5, "output_tokens": 7},
|
||
},
|
||
}
|
||
return ("data: " + json.dumps(载荷, ensure_ascii=False) + "\n\n").encode()
|
||
|
||
class 请求处理(BaseHTTPRequestHandler):
|
||
def do_POST(self):
|
||
try:
|
||
json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
||
输出 = 回应({})
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/event-stream")
|
||
self.send_header("Content-Length", str(len(输出)))
|
||
self.end_headers()
|
||
self.wfile.write(输出)
|
||
except Exception:
|
||
self.send_error(500)
|
||
|
||
def log_message(self, *args):
|
||
pass
|
||
|
||
from muse.编排.生成正文 import 处理器版本集 as 生成处理器集
|
||
from muse.编排.章后处理 import 处理器版本集 as 章后处理器集
|
||
|
||
server = ThreadingHTTPServer(("127.0.0.1", 0), 请求处理)
|
||
thread = Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
|
||
凭据 = tmp_path / "post-key"
|
||
凭据.write_text("synthetic-derivation-credential")
|
||
原配置 = 运行配置内容(
|
||
"direct",
|
||
"1",
|
||
策略.定义["version"],
|
||
策略.资源发布身份,
|
||
"synthetic",
|
||
{
|
||
"extractor": {
|
||
"provider": "synthetic",
|
||
"model": "claude-opus-4-8[1M]",
|
||
"thinking": "high",
|
||
"stage_tools": {"抽取": []},
|
||
}
|
||
},
|
||
(凭据引用("provider-key", "受控存储", str(凭据)),),
|
||
(
|
||
提供方配置(
|
||
"synthetic",
|
||
"responses",
|
||
f"http://127.0.0.1:{server.server_port}/v1/responses",
|
||
"provider-key",
|
||
),
|
||
),
|
||
合成派生计价().版本,
|
||
)
|
||
管理 = 配置版本管理(库, _离线派生配置检查())
|
||
管理.保存草案("post-config", "1", 原配置)
|
||
启用回执 = 管理.验证版本("post-config", "1")
|
||
管理.启用("post-config", "1", 验证回执=启用回执, 批准引用="synthetic-approval", 预期代次=0)
|
||
预算 = 预算管理(库, "synthetic")
|
||
预算.登记策略(额度策略("synthetic", "1", Decimal("20"), 40))
|
||
|
||
作者 = 调用身份("post-author", None, 库.用途, 内容用途.抽取)
|
||
作品 = 装配.要求作品()
|
||
|
||
def 新作品(work_id: str, 章数: int = 2, 章前缀: str = "ch-") -> None:
|
||
结构选择 = 作品.新档案结构(作者, work_id, "work_core", 1)
|
||
作品.保存档案(
|
||
作者,
|
||
f"create-{work_id}",
|
||
档案保存(
|
||
work_id,
|
||
0,
|
||
{"名称": work_id, "题材定位": "现实", "主题立意": "守约"},
|
||
结构选择,
|
||
),
|
||
)
|
||
for i in range(1, 章数 + 1):
|
||
作品.添加章节(
|
||
作者,
|
||
f"add-{work_id}-{章前缀}{i}",
|
||
work_id,
|
||
f"{章前缀}{i}",
|
||
f"第{i}章",
|
||
预期目录版本=i - 1,
|
||
)
|
||
|
||
新作品("post-work", 3)
|
||
|
||
环境 = {
|
||
"库": 库,
|
||
"装配": 装配,
|
||
"作者": 作者,
|
||
"剧本": 剧本,
|
||
"收到": 收到,
|
||
"新作品": 新作品,
|
||
"提供方端口": server.server_port,
|
||
"处理器能力": sorted(set(生成处理器集) | set(章后处理器集)),
|
||
}
|
||
yield 环境
|
||
server.shutdown()
|
||
server.server_close()
|
||
thread.join(timeout=5)
|
||
|
||
|
||
@pytest.fixture
|
||
def 章后工具(章后环境):
|
||
"""章后旅程的具名操作：保存一章、驱动任务、构造抽取剧本。"""
|
||
|
||
from types import SimpleNamespace
|
||
from uuid import uuid4
|
||
|
||
from muse.任务运行.接口 import 任务状态
|
||
from muse.正文写作.接口 import 文本节点, 正文草稿, 段落
|
||
|
||
环境 = 章后环境
|
||
|
||
def 保存一章(章: str, 文本: str, 命令: str | None = None, 预期版本: int = 0):
|
||
回执 = (
|
||
环境["装配"]
|
||
.要求正文()
|
||
.保存人工(
|
||
环境["作者"],
|
||
命令 or f"save-{章}-{uuid4().hex[:6]}",
|
||
章,
|
||
预期版本,
|
||
正文草稿((段落(f"{章}-p1", (文本节点(文本),)),)),
|
||
)
|
||
)
|
||
来源 = next(
|
||
(r for r in 回执["results"] if isinstance(r, dict) and r.get("章后任务")),
|
||
None,
|
||
)
|
||
assert 来源 is not None, f"章后任务未登记：{回执['results']}"
|
||
return 来源["章后任务"], 回执
|
||
|
||
def 跑任务(任务ID: str):
|
||
运行 = 环境["装配"].任务运行
|
||
while True:
|
||
快照 = 运行.读取任务(任务ID)
|
||
if 快照.状态 in {任务状态.已完成, 任务状态.已失败, 任务状态.已取消}:
|
||
return 快照
|
||
领取 = 运行.领取步骤("worker", 环境["处理器能力"])
|
||
assert 领取 is not None, "任务未完成却没有可领取步骤"
|
||
运行.执行一步(领取)
|
||
|
||
def 抽取输出(摘要: str, 事实: list[dict]) -> dict:
|
||
return {"类型": "文本", "文本": {"summary": 摘要, "facts": 事实}}
|
||
|
||
return SimpleNamespace(保存一章=保存一章, 跑任务=跑任务, 抽取输出=抽取输出, 环境=环境)
|
||
|
||
|
||
@pytest.fixture
|
||
def 研究环境(章后环境, tmp_path):
|
||
"""研究拆书环境：analyst 配置、研究流程装配与带授权的参考来源。"""
|
||
|
||
from decimal import Decimal
|
||
from types import SimpleNamespace
|
||
from uuid import uuid4
|
||
|
||
from muse.任务运行.配置版本 import 凭据引用, 提供方配置, 运行配置内容, 配置版本管理
|
||
from muse.任务运行.预算管理 import 预算管理, 额度策略
|
||
from muse.启动 import 装配研究拆书
|
||
from muse.编排.研究拆书 import 处理器版本集 as 研究处理器集
|
||
from muse.资料研究.接口 import 资料服务
|
||
from muse.资料研究.模型 import 导入请求
|
||
|
||
环境 = 章后环境
|
||
环境["处理器能力"] = sorted(set(环境["处理器能力"]) | set(研究处理器集))
|
||
库, 装配 = 环境["库"], 环境["装配"]
|
||
策略 = __import__("muse.任务运行.角色策略", fromlist=["角色策略目录"]).角色策略目录.从发布包()
|
||
|
||
装配研究拆书(装配, 合成派生计价(), 分析配置ID="research-config")
|
||
|
||
凭据 = tmp_path / "research-key"
|
||
凭据.write_text("synthetic-research-credential")
|
||
凭据.chmod(0o600)
|
||
研究配置 = 运行配置内容(
|
||
"direct",
|
||
"1",
|
||
策略.定义["version"],
|
||
策略.资源发布身份,
|
||
"synthetic",
|
||
{
|
||
"extractor": {
|
||
"provider": "synthetic",
|
||
"model": "claude-opus-4-8[1M]",
|
||
"thinking": "high",
|
||
"stage_tools": {"抽取": [], "拆书分析": []},
|
||
},
|
||
},
|
||
(凭据引用("provider-key", "受控存储", str(凭据)),),
|
||
(
|
||
提供方配置(
|
||
"synthetic",
|
||
"responses",
|
||
f"http://127.0.0.1:{环境['提供方端口']}/v1/responses",
|
||
"provider-key",
|
||
),
|
||
),
|
||
合成派生计价().版本,
|
||
)
|
||
管理 = 配置版本管理(库, _离线派生配置检查())
|
||
管理.保存草案("research-config", "1", 研究配置)
|
||
启用回执 = 管理.验证版本("research-config", "1")
|
||
管理.启用("research-config", "1", 验证回执=启用回执, 批准引用="synthetic-approval", 预期代次=0)
|
||
预算 = 预算管理(库, "synthetic-research")
|
||
预算.登记策略(额度策略("synthetic-research", "1", Decimal("40"), 80))
|
||
|
||
服务 = 资料服务()
|
||
|
||
def 导入来源(标题: str, 正文: str, 用途集: tuple[str, ...] = ("analysis",)) -> dict:
|
||
with 库.连接() as 连, 连.transaction():
|
||
回执 = 服务.导入(
|
||
连,
|
||
环境["作者"].作者,
|
||
导入请求("reference", 标题, f"file://资料/{标题}.txt", 正文, 用途集),
|
||
)
|
||
return {
|
||
"source_id": 回执.source_id,
|
||
"revision": 回执.revision,
|
||
"content_hash": 回执.content_hash,
|
||
}
|
||
|
||
def 发起(
|
||
source_id: str, *, 配置ID: str = "research-config", 重跑窗=None, 理由=None, 窗口章数=None
|
||
):
|
||
from muse.编排.研究拆书 import 发起研究拆书
|
||
|
||
样本 = {"理由": 理由} if 理由 else None
|
||
return 发起研究拆书(
|
||
装配,
|
||
环境["作者"],
|
||
"research-" + uuid4().hex[:8],
|
||
source_id=source_id,
|
||
配置ID=配置ID,
|
||
重跑窗=重跑窗,
|
||
样本依据=样本,
|
||
窗口章数=窗口章数,
|
||
)
|
||
|
||
def 跑任务(任务ID: str):
|
||
from muse.任务运行.接口 import 任务状态
|
||
|
||
运行 = 装配.任务运行
|
||
能力 = sorted(set(环境["处理器能力"]) | set(研究处理器集))
|
||
while True:
|
||
快照 = 运行.读取任务(任务ID)
|
||
if 快照.状态 in {任务状态.已完成, 任务状态.已失败, 任务状态.已取消}:
|
||
return 快照
|
||
领取 = 运行.领取步骤("worker", 能力)
|
||
assert 领取 is not None, "任务未完成却没有可领取步骤"
|
||
运行.执行一步(领取)
|
||
|
||
def 分析输出(实体: list[dict], 写法=("三段式推进",)) -> dict:
|
||
return {
|
||
"类型": "文本",
|
||
"文本": {
|
||
"chapters": [{"title": "章", "beat": "节拍"}],
|
||
"phases": ["推进"],
|
||
"entities": 实体,
|
||
"techniques": list(写法),
|
||
},
|
||
}
|
||
|
||
return SimpleNamespace(
|
||
环境=环境,
|
||
库=库,
|
||
装配=装配,
|
||
剧本=环境["剧本"],
|
||
服务=服务,
|
||
导入来源=导入来源,
|
||
发起=发起,
|
||
跑任务=跑任务,
|
||
分析输出=分析输出,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def 新空目标库(应用测试库: dict, 隔离数据库URL: str, tmp_path: Path):
|
||
import uuid
|
||
from types import SimpleNamespace
|
||
|
||
import psycopg
|
||
from psycopg import sql
|
||
from psycopg.conninfo import conninfo_to_dict, make_conninfo
|
||
|
||
from muse.共享.调用身份 import 用途
|
||
from muse.基础设施.数据库.连接 import 数据库工厂
|
||
from muse.配置 import 数据库引用
|
||
|
||
assert 应用测试库[用途.维护].用途 is 用途.维护
|
||
库名 = "muse_w29_restore_" + uuid.uuid4().hex[:12]
|
||
with psycopg.connect(隔离数据库URL, autocommit=True) as 管理:
|
||
管理.execute(sql.SQL("CREATE DATABASE {} OWNER muse_maint").format(sql.Identifier(库名)))
|
||
参数 = conninfo_to_dict(隔离数据库URL)
|
||
参数.update(dbname=库名, user="muse_maint")
|
||
引用文件 = tmp_path / f"{库名}.txt"
|
||
引用文件.write_text(make_conninfo(**参数), encoding="utf-8")
|
||
引用文件.chmod(0o600)
|
||
工厂 = 数据库工厂(数据库引用("受控存储", str(引用文件)), 用途.维护)
|
||
try:
|
||
yield SimpleNamespace(库名=库名, 工厂=工厂)
|
||
finally:
|
||
with psycopg.connect(隔离数据库URL, autocommit=True) as 管理:
|
||
管理.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(库名)))
|