d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""真实写手发送、快照与HTTP决定，验证偏好撤回的后续影响。"""
|
||
|
||
import json
|
||
import os
|
||
from dataclasses import asdict, replace
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
import test_两阶段写手 as 写手测试
|
||
import test_作者偏好与反馈 as 偏好测试
|
||
import test_写手声音消费 as 声音测试
|
||
import test_受控规划方法消费 as 规划测试
|
||
import test_生产评测权限隔离 as 接入测试
|
||
from fastapi.testclient import TestClient
|
||
|
||
from muse.上下文.接口 import 核对依赖
|
||
from muse.作者经验.接口 import 偏好决定, 偏好范围, 偏好请求
|
||
from muse.接入.http.应用 import 创建应用
|
||
from muse.编排.生成正文 import 生成正文错误
|
||
from muse.配置 import 读取配置
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
生成环境 = 写手测试.生成环境
|
||
候选环境 = 偏好测试.候选环境
|
||
环境 = 偏好测试.环境
|
||
规划环境 = 规划测试.规划环境
|
||
|
||
|
||
def _偏好(env):
|
||
req = 偏好请求(
|
||
str(uuid4()),
|
||
"心理描写",
|
||
"以动作写心理，不替人物解释。",
|
||
"以后都按这个偏好写。",
|
||
偏好范围("work", "gen-work"),
|
||
)
|
||
saved = env["装配"].要求作者经验().保存偏好(env["作者"], str(uuid4()), req)["results"][0]
|
||
return 偏好决定(saved["preference_id"], saved["revision"], saved["content_hash"])
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w26-26b001",
|
||
environment="隔离PG；实际S01/B05/B09/S02与合成HTTP，不证明文学效果",
|
||
given="真实作者、作品及确切正文反馈版本",
|
||
when="保存范围偏好、审阅冲突、确认、撤回及重新消费",
|
||
then=["作者偏好进入实际写手且撤回只影响后续"],
|
||
contract="docs/系统架构/新版设计/模块设计/B07-作者经验.md",
|
||
)
|
||
def test_作者偏好进入实际写手且撤回只影响后续__26b001(生成环境):
|
||
env = 生成环境
|
||
pref = _偏好(env)
|
||
_, task = 写手测试.发起并跑完(env, 剧本=写手测试.正常剧本(["林深停在码头。脚边一封湿信。 "]))
|
||
snapshot = 声音测试.读取材料(env, task)
|
||
assert "以动作写心理" in json.loads(env["收到"][-1]["input"])["装配材料"]
|
||
assert snapshot.冻结范围["preferences"][0]["preference_id"] == pref.preference_id
|
||
env["装配"].要求作者经验().撤回偏好(env["作者"], "withdraw", pref)
|
||
with env["库"].连接(只读=True) as conn:
|
||
assert not 核对依赖(conn, snapshot.冻结范围).有效
|
||
historical = env["装配"].要求上下文().读取快照(snapshot.快照ID)
|
||
assert historical.装配["文本"] == snapshot.装配["文本"]
|
||
_, next_task = 写手测试.发起并跑完(env, 剧本=写手测试.正常剧本(["门扇开了。 "]))
|
||
assert 声音测试.读取材料(env, next_task).冻结范围["preferences"] == []
|
||
assert "以动作写心理" not in json.loads(env["收到"][-1]["input"])["装配材料"]
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w26-26b002",
|
||
environment="隔离PG；实际S01/B05/B09/S02与合成HTTP，不证明文学效果",
|
||
given="真实作者、作品及确切正文反馈版本",
|
||
when="保存范围偏好、审阅冲突、确认、撤回及重新消费",
|
||
then=["冻结后新增或撤回偏好拒绝旧写手发送"],
|
||
contract="docs/系统架构/新版设计/模块设计/B07-作者经验.md",
|
||
)
|
||
@pytest.mark.parametrize("已有", [False, True])
|
||
def test_冻结后新增或撤回偏好拒绝旧写手发送__26b002(生成环境, 已有):
|
||
env = 生成环境
|
||
pref = _偏好(env) if 已有 else None
|
||
|
||
def 改变(tid):
|
||
if pref:
|
||
env["装配"].要求作者经验().撤回偏好(env["作者"], "withdraw", pref)
|
||
else:
|
||
_偏好(env)
|
||
|
||
with pytest.raises(生成正文错误, match="偏好"):
|
||
写手测试.发起并跑完(env, 剧本=写手测试.正常剧本(["林深停在码头。"]), 冻结后=改变)
|
||
assert len(env["收到"]) == 4
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w26-26b004",
|
||
environment="隔离PG；实际S01/B05/B09/S02与合成HTTP，不证明文学效果",
|
||
given="真实作者、作品及确切正文反馈版本",
|
||
when="保存范围偏好、审阅冲突、确认、撤回及重新消费",
|
||
then=["同一偏好进入实际规划请求"],
|
||
contract="docs/系统架构/新版设计/模块设计/B07-作者经验.md",
|
||
)
|
||
def test_同一偏好进入实际规划请求__26b004(规划环境):
|
||
env = 规划环境
|
||
pref = _偏好(env)
|
||
_, task = 规划测试._运行(env)
|
||
assert task.状态.value == "completed", task.步骤
|
||
assert "以动作写心理" in json.loads(env["收到"][-1]["input"])["装配材料"]
|
||
assert any(
|
||
x["preference_id"] == pref.preference_id
|
||
for x in task.冻结输入["冻结上下文"]["source_scope"]["preferences"]
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w26-26b003",
|
||
environment="隔离PG；实际S01/B05/B09/S02与合成HTTP，不证明文学效果",
|
||
given="真实作者、作品及确切正文反馈版本",
|
||
when="保存范围偏好、审阅冲突、确认、撤回及重新消费",
|
||
then=["HTTP反馈到偏好审阅与撤回真实读回"],
|
||
contract="docs/系统架构/新版设计/模块设计/B07-作者经验.md",
|
||
)
|
||
def test_HTTP反馈到偏好审阅与撤回真实读回__26b003(环境, tmp_path):
|
||
service, author, pools, feedback = 环境
|
||
config = 接入测试._配置文件(pools[author.用途], tmp_path)
|
||
config.write_text(config.read_text().replace("eval-author", author.作者))
|
||
with TestClient(创建应用(读取配置(config)), headers={"origin": "http://testserver"}) as client:
|
||
assert client.get("/api/v1/preferences").status_code == 401
|
||
assert (
|
||
client.post(
|
||
"/api/v1/session", json={"password": "synthetic-evaluation-only"}
|
||
).status_code
|
||
== 200
|
||
)
|
||
res = client.post(
|
||
"/api/v1/feedback", json={"command_id": "http-feedback", "feedback": asdict(feedback)}
|
||
)
|
||
assert res.status_code == 200, res.text
|
||
fid = res.json()["feedback_id"]
|
||
assert "改过的首句" in client.get("/api/v1/feedback/" + fid).json()["after_text"]
|
||
req = 偏好测试._请求(origin="inferred", feedback_id=fid, quote=feedback.quote)
|
||
res = client.post(
|
||
"/api/v1/preferences", json={"command_id": "save", "preference": asdict(req)}
|
||
)
|
||
assert res.status_code == 200, res.text
|
||
saved = res.json()["results"][0]
|
||
point = asdict(偏好测试._定位(saved))
|
||
review = client.post("/api/v1/preferences/review", json={"selection": point})
|
||
assert review.status_code == 200, review.text
|
||
res = client.post(
|
||
"/api/v1/preferences/confirm",
|
||
json={
|
||
"command_id": "confirm",
|
||
"selection": point,
|
||
"review_id": review.json()["review_id"],
|
||
"review_hash": review.json()["review_hash"],
|
||
},
|
||
)
|
||
assert res.status_code == 200, res.text
|
||
active = res.json()["results"][0]
|
||
assert active["state"] == "active"
|
||
res = client.post(
|
||
"/api/v1/preferences/revoke",
|
||
json={"command_id": "revoke", "selection": asdict(偏好测试._定位(active))},
|
||
)
|
||
assert res.status_code == 200, res.text
|
||
assert client.get("/api/v1/preferences").json()[0]["state"] == "withdrawn"
|
||
if os.environ.get("MUSE_RUN_BROWSER") == "1":
|
||
_浏览器偏好(读取配置(config), fid, tmp_path)
|
||
rows = service.列出偏好(author)
|
||
assert len(rows) == 2 and all(r["state"] == "withdrawn" for r in rows)
|
||
|
||
|
||
def _浏览器偏好(config, feedback_id, tmp_path, journey="作者偏好.spec.ts"):
|
||
import socket
|
||
import subprocess
|
||
from contextlib import asynccontextmanager
|
||
from pathlib import Path
|
||
from threading import Event, Thread
|
||
|
||
import uvicorn
|
||
|
||
from muse.配置 import 服务配置
|
||
|
||
password = tmp_path / "preference-browser-key"
|
||
password.write_text("synthetic-preference-only")
|
||
password.chmod(0o600)
|
||
sock = socket.socket()
|
||
sock.bind(("127.0.0.1", 0))
|
||
port = sock.getsockname()[1]
|
||
base = f"http://127.0.0.1:{port}"
|
||
app = 创建应用(
|
||
replace(
|
||
config,
|
||
HTTP=服务配置(
|
||
str(password), 作者ID=config.HTTP.作者ID, 端口=port, 公开地址=base, 允许来源=(base,)
|
||
),
|
||
)
|
||
)
|
||
ready = Event()
|
||
previous = app.router.lifespan_context
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(instance):
|
||
async with previous(instance):
|
||
ready.set()
|
||
yield
|
||
|
||
app.router.lifespan_context = lifespan
|
||
server = uvicorn.Server(
|
||
uvicorn.Config(app, log_level="warning", access_log=False, timeout_graceful_shutdown=3)
|
||
)
|
||
thread = Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
|
||
output = Path(os.environ.get("MUSE_BROWSER_ARTIFACTS", str(tmp_path / "browser"))).resolve()
|
||
output.mkdir(parents=True, exist_ok=True)
|
||
env = {
|
||
**os.environ,
|
||
"MUSE_WORKBENCH_URL": base,
|
||
"MUSE_AUTHOR_PASSWORD_FILE": str(password),
|
||
"MUSE_FEEDBACK_ID": feedback_id,
|
||
}
|
||
chrome = Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome")
|
||
if chrome.exists():
|
||
env.setdefault("MUSE_BROWSER_EXECUTABLE", str(chrome))
|
||
thread.start()
|
||
try:
|
||
assert ready.wait(10)
|
||
with (output / "浏览器.json").open("w") as log:
|
||
process = subprocess.run(
|
||
[
|
||
"pnpm",
|
||
"exec",
|
||
"playwright",
|
||
"test",
|
||
journey,
|
||
"--reporter=json",
|
||
"--output",
|
||
str(output / "产物"),
|
||
],
|
||
cwd=Path(__file__).parents[2] / "web",
|
||
env=env,
|
||
stdout=log,
|
||
stderr=subprocess.STDOUT,
|
||
timeout=90,
|
||
)
|
||
assert process.returncode == 0, str(output / "浏览器.json")
|
||
return output
|
||
finally:
|
||
server.should_exit = True
|
||
thread.join(10)
|
||
sock.close()
|
||
assert not thread.is_alive()
|