test(e2e): 补上打真实网关的那一层,四层测试到齐
两个用例:一次运行走通「模型→解释→工具→观察回填→再问模型→收尾」,以及取消能穿过真实的 HTTP 请求并留下已取消的结束记录。测试自带解释器、工具注册表与事件出口——库故意不带它们, 带了就等于替某一家定了动作语言。 断言只绑结构不变量,一条都不绑模型输出的文字:停止原因落在两个正常终态里、至少两步且至少 一步真的执行过动作、每步 call_id 非空(网关那边生成的,替身给不出来)、日志读回来与返回值 逐字段相等、事件条数等于步数。绑内容的测试会随机红,而随机红的测试很快没人看。 两道跳过闸:网关装没装,以及 POLYLOOP_E2E 是不是 1。分开是因为填好密钥不等于同意花钱。 实测跑一次三次真实模型调用、约 12 秒;开关关着时整层跳过;make ci 283 passed / 16 skipped。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -39,9 +39,10 @@
|
|||||||
所以它排在架构前面:边界画错,后面每一份架构文档都要重写
|
所以它排在架构前面:边界画错,后面每一份架构文档都要重写
|
||||||
- [x] ③ 架构 —— `research-wiki/explanation/architecture.md` 与 `pyproject.toml` 的 import-linter 契约。
|
- [x] ③ 架构 —— `research-wiki/explanation/architecture.md` 与 `pyproject.toml` 的 import-linter 契约。
|
||||||
架构文档先于代码存在,此期间它是一份规格而不是描述,文档开头须写明这一点
|
架构文档先于代码存在,此期间它是一份规格而不是描述,文档开头须写明这一点
|
||||||
- [x] ④ 测试框架 —— unit / integration / contract 三层已在跑;**e2e 还是空的**,它要打真实
|
- [x] ④ 测试框架 —— 四层都在跑(划分判据是「依赖什么」,见 [CLAUDE.md](CLAUDE.md) §1.9)。
|
||||||
模型网关(划分判据是「依赖什么」,见 [CLAUDE.md](CLAUDE.md) §1.9)。`tests/contract/`
|
e2e 打真实模型网关、会产生真实费用,所以默认不跑:要 `POLYLOOP_E2E=1` 加显式
|
||||||
那套公共行为一致性用例接上了自带的存储实现,解释器与执行器那几条仍等下游把实现接进来
|
`pytest -m e2e`,配置见 [.env.example](.env.example)。`tests/contract/` 那套公共行为
|
||||||
|
一致性用例接上了自带的存储实现,解释器与执行器那几条仍等下游把实现接进来
|
||||||
- [x] ⑤ 实现 —— 十个模块全部落地,五个接缝都有调用点。**一处已知欠账**:`stores` 只有逐行
|
- [x] ⑤ 实现 —— 十个模块全部落地,五个接缝都有调用点。**一处已知欠账**:`stores` 只有逐行
|
||||||
追加那一种形态,关系数据库那种由下游自己实现,契约套件是它的准入标准
|
追加那一种形态,关系数据库那种由下游自己实现,契约套件是它的准入标准
|
||||||
- [ ] ⑥ 迁移验收 —— 真的把 dissect 与 GovDoc-SaaS 迁过来,以两边测试全绿为准
|
- [ ] ⑥ 迁移验收 —— 真的把 dissect 与 GovDoc-SaaS 迁过来,以两边测试全绿为准
|
||||||
|
|||||||
@@ -0,0 +1,421 @@
|
|||||||
|
"""打真实模型网关的那一层:一次运行真的从模型走到工具再走回来。
|
||||||
|
|
||||||
|
**这一层叫 e2e,因为它连的是真实模型网关**(`CLAUDE.md` §1.9 的分层判据是「依赖什么」)。
|
||||||
|
它会产生真实的模型调用与真实的费用,所以它是唯一一层默认不跑的测试——`pyproject.toml` 的
|
||||||
|
`addopts` 里有 `-m 'not e2e'`,`make ci` 因此跑不到这里。要跑它得显式写 `pytest -m e2e`。
|
||||||
|
|
||||||
|
**两道跳过闸,缺一不可。** 第一道是网关装没装(没装 `polyloop[gateway]` 就整份文件跳过);
|
||||||
|
第二道是 `POLYLOOP_E2E` 这个开关等不等于 `"1"`。分成两件事是因为**填好密钥不等于同意花钱**:
|
||||||
|
只看密钥的话,任何人配好 `.env` 之后随手跑一次全套测试就会打出去一串真实调用并产生真实账单,
|
||||||
|
而他本来只是想看看测试过不过。开关的读法与网关一致——先读当前工作目录下的 `.env`,再让环境
|
||||||
|
变量覆盖它,所以临时开一次不必改文件。
|
||||||
|
|
||||||
|
**这里自带三样真实实现:决策解释器、工具注册表、事件出口。** 库故意不带它们(带了就等于替
|
||||||
|
某一家定了动作语言),而没有它们循环就走不起来。它们住在测试里,不是库的一部分。
|
||||||
|
|
||||||
|
**断言只绑结构不变量,一条都不绑模型输出的文字内容。** 模型是不确定的,绑内容的测试会随机
|
||||||
|
红,而随机红的测试很快就会被所有人忽略,然后这一层就不再拦得住任何东西。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
from collections.abc import Mapping
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
polygateway = pytest.importorskip(
|
||||||
|
"polygateway", reason="没装 polyloop[gateway],打真实网关这一层跳过"
|
||||||
|
)
|
||||||
|
|
||||||
|
from dotenv import dotenv_values # noqa: E402
|
||||||
|
from polygateway import GatewayClient, GatewaySettings # noqa: E402
|
||||||
|
|
||||||
|
from polyloop.adapters import GatewayModelClient # noqa: E402
|
||||||
|
from polyloop.ports import ( # noqa: E402
|
||||||
|
Action,
|
||||||
|
Event,
|
||||||
|
FinalAnswer,
|
||||||
|
InvalidDecision,
|
||||||
|
ModelCall,
|
||||||
|
ParsedReply,
|
||||||
|
ToolCall,
|
||||||
|
)
|
||||||
|
from polyloop.session import AgentDefinition, RunRequest, run # noqa: E402
|
||||||
|
from polyloop.stores import JsonlRunStore # noqa: E402
|
||||||
|
from polyloop.tools import ToolRegistry, ToolSpec # noqa: E402
|
||||||
|
from polyloop.types import ( # noqa: E402
|
||||||
|
ActionStatus,
|
||||||
|
Budget,
|
||||||
|
Context,
|
||||||
|
Message,
|
||||||
|
ModelReply,
|
||||||
|
ReplayPolicy,
|
||||||
|
Role,
|
||||||
|
StopReason,
|
||||||
|
SyntheticObservations,
|
||||||
|
TextBlock,
|
||||||
|
)
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.e2e
|
||||||
|
|
||||||
|
#: 开关的读法与网关一致:`.env` 在下、环境变量在上。两份都读是因为密钥本来就在 `.env` 里,
|
||||||
|
#: 而临时开一次 e2e 不该逼人去改那个文件。
|
||||||
|
_ENV = {**dotenv_values(".env"), **os.environ}
|
||||||
|
|
||||||
|
if _ENV.get("POLYLOOP_E2E") != "1":
|
||||||
|
pytest.skip(
|
||||||
|
"POLYLOOP_E2E 不是 1:这一层会打真实模型网关并产生真实费用,默认不跑",
|
||||||
|
allow_module_level=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 动作协议:模型每一轮只输出一个 JSON 对象
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
#: 讲给模型听的那份协议。**写得极其明确并给一个例子**——模型只有照这个格式输出,循环才走得
|
||||||
|
#: 下去。指望它猜的话,第一轮就会得到一段散文,然后这次运行以连续解析失败收尾,而那个红叉
|
||||||
|
#: 看起来像是库坏了。
|
||||||
|
_PROTOCOL = """你在一个自动循环里工作。每一轮你**只能输出一个 JSON 对象**,前后不许有任何别的文字、说明或标点。
|
||||||
|
|
||||||
|
要调用工具,输出:
|
||||||
|
{"tool": "工具名", "args": {"参数名": 参数值}}
|
||||||
|
|
||||||
|
要给出最终回答,输出:
|
||||||
|
{"final": "你的回答"}
|
||||||
|
|
||||||
|
可用的工具只有两个:
|
||||||
|
- add:把两个数相加。参数 a 和 b 都是数字。
|
||||||
|
- submit:提交结果。参数 answer 是一个字符串。调用它就表示这次工作做完了。
|
||||||
|
|
||||||
|
例子——要算 3 加 4,你这一轮就输出:
|
||||||
|
{"tool": "add", "args": {"a": 3, "b": 4}}
|
||||||
|
|
||||||
|
每一轮之后你会收到一条以「观察:」开头的消息,那是上一次工具调用返回的内容。"""
|
||||||
|
|
||||||
|
_GOAL = "请先用 add 算出 17 加 25,拿到结果之后用 submit 把那个结果提交上去。"
|
||||||
|
|
||||||
|
_OBSERVATION_TEMPLATE = "观察:{observation}"
|
||||||
|
|
||||||
|
#: 代码围栏。模型很常把 JSON 包在 ```json ... ``` 里,剥不掉的话每一轮都会解析失败。
|
||||||
|
_FENCE = re.compile(r"```[A-Za-z0-9_+-]*\n(?P<body>.*?)```", re.DOTALL)
|
||||||
|
|
||||||
|
|
||||||
|
def _json_payload(text: str) -> str:
|
||||||
|
"""把模型这一轮的输出削到只剩那个 JSON 对象。
|
||||||
|
|
||||||
|
两步都是必要的:先剥围栏,再取最外层花括号之间的那一段。只剥围栏的话,模型在 JSON 前后
|
||||||
|
写一句「好的,我来算一下」就解析不了;只取花括号的话,围栏里带语言标签的那种输出会把
|
||||||
|
```json 一起吃进去。围栏没有闭合时第一步不匹配,第二步照样能把 JSON 捞出来。
|
||||||
|
"""
|
||||||
|
body = text.strip()
|
||||||
|
fenced = _FENCE.search(body)
|
||||||
|
if fenced is not None:
|
||||||
|
body = fenced.group("body")
|
||||||
|
start = body.find("{")
|
||||||
|
end = body.rfind("}")
|
||||||
|
if start != -1 and end > start:
|
||||||
|
body = body[start : end + 1]
|
||||||
|
return body.strip()
|
||||||
|
|
||||||
|
|
||||||
|
class _JsonDecisionParser:
|
||||||
|
"""按上面那份协议解释一次模型回复。
|
||||||
|
|
||||||
|
**对任何输入都返回 `ParsedReply`,绝不抛异常**(`design/0007` 决策三)。解释不出来是正常
|
||||||
|
路径的一部分——模型输出不合格式是每天都在发生的事,而抛异常会让库去替它编一个停止原因,
|
||||||
|
于是「解释器有 bug」被伪装成「这次运行以某某原因结束」,然后进下游的统计。
|
||||||
|
|
||||||
|
每一种失败给一条**对症**的说明,因为那段文本就是回喂给模型的观察。压成一句「格式错误」
|
||||||
|
的话,模型不知道自己错在哪,下一轮多半照错一遍。
|
||||||
|
"""
|
||||||
|
|
||||||
|
def parse(self, reply: ModelReply) -> ParsedReply:
|
||||||
|
payload = _json_payload(reply.content)
|
||||||
|
try:
|
||||||
|
decoded = json.loads(payload)
|
||||||
|
except json.JSONDecodeError as exc:
|
||||||
|
return self._invalid(
|
||||||
|
reply,
|
||||||
|
f"这一轮的输出不是一个 JSON 对象({exc.msg})。"
|
||||||
|
'只输出一个 JSON 对象,形如 {"tool": "add", "args": {"a": 1, "b": 2}},前后不要有别的文字。',
|
||||||
|
)
|
||||||
|
if not isinstance(decoded, dict):
|
||||||
|
return self._invalid(
|
||||||
|
reply,
|
||||||
|
f"解出来的是 {type(decoded).__name__} 而不是一个 JSON 对象。"
|
||||||
|
'只输出一个 JSON 对象,形如 {"tool": "add", "args": {"a": 1, "b": 2}}。',
|
||||||
|
)
|
||||||
|
if "final" in decoded:
|
||||||
|
return ParsedReply(
|
||||||
|
history_text=reply.content, decision=FinalAnswer(text=str(decoded["final"]))
|
||||||
|
)
|
||||||
|
if "tool" not in decoded:
|
||||||
|
return self._invalid(
|
||||||
|
reply,
|
||||||
|
'这个 JSON 对象里既没有 "tool" 也没有 "final"。调工具用 '
|
||||||
|
'{"tool": ..., "args": {...}},给最终回答用 {"final": "..."}。',
|
||||||
|
)
|
||||||
|
name = decoded["tool"]
|
||||||
|
if not isinstance(name, str):
|
||||||
|
return self._invalid(reply, '"tool" 必须是一个字符串,也就是工具的名字。')
|
||||||
|
arguments = decoded.get("args", {})
|
||||||
|
if not isinstance(arguments, dict):
|
||||||
|
return self._invalid(
|
||||||
|
reply, '"args" 必须是一个 JSON 对象,键是参数名,例如 {"a": 1, "b": 2}。'
|
||||||
|
)
|
||||||
|
return ParsedReply(
|
||||||
|
history_text=reply.content,
|
||||||
|
decision=Action(
|
||||||
|
text=f"{name}({json.dumps(arguments, ensure_ascii=False, sort_keys=True)})",
|
||||||
|
tool_call=ToolCall(name=name, arguments=arguments),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
def parameters(self) -> Mapping[str, str]:
|
||||||
|
return {"kind": "json-tool-or-final"}
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _invalid(reply: ModelReply, explanation: str) -> ParsedReply:
|
||||||
|
return ParsedReply(
|
||||||
|
history_text=reply.content, decision=InvalidDecision(explanation=explanation)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 工具:纯计算,不碰文件系统、网络、子进程
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
async def _add(arguments: Mapping[str, object]) -> str:
|
||||||
|
"""两数相加。
|
||||||
|
|
||||||
|
注册表已经按 schema 校验过类型,这里仍然自己判一次:校验的是 JSON Schema 的一个子集,
|
||||||
|
而一个 `TypeError` 从这里抛出去会被判成「已执行」加一条正常观察,模型看不出该怎么改。
|
||||||
|
"""
|
||||||
|
a, b = arguments.get("a"), arguments.get("b")
|
||||||
|
if not isinstance(a, int | float) or not isinstance(b, int | float):
|
||||||
|
return "a 和 b 都必须是数字。"
|
||||||
|
return str(a + b)
|
||||||
|
|
||||||
|
|
||||||
|
async def _submit(arguments: Mapping[str, object]) -> str:
|
||||||
|
return f"收到:{arguments.get('answer')}"
|
||||||
|
|
||||||
|
|
||||||
|
def _registry() -> ToolRegistry:
|
||||||
|
"""本次运行可见的两个工具。
|
||||||
|
|
||||||
|
`submit` 带完成标记,这样运行有一条确定的收尾路径——没有它的话,这次运行只能靠模型自己
|
||||||
|
给最终回答或者撞上步数上限收尾,而那两条路一条不确定、一条要多花几次调用。
|
||||||
|
"""
|
||||||
|
return ToolRegistry(
|
||||||
|
(
|
||||||
|
ToolSpec(
|
||||||
|
name="add",
|
||||||
|
description="把两个数相加,返回它们的和。",
|
||||||
|
parameters={
|
||||||
|
"type": "object",
|
||||||
|
"properties": {"a": {"type": "number"}, "b": {"type": "number"}},
|
||||||
|
"required": ["a", "b"],
|
||||||
|
"additionalProperties": False,
|
||||||
|
},
|
||||||
|
# 纯计算,重复算一次无害。
|
||||||
|
replay_policy=ReplayPolicy.SAFE,
|
||||||
|
handler=_add,
|
||||||
|
),
|
||||||
|
ToolSpec(
|
||||||
|
name="submit",
|
||||||
|
description="提交最终结果。调用它就表示这次工作做完了。",
|
||||||
|
parameters={
|
||||||
|
"type": "object",
|
||||||
|
# 允许数字:模型很常把算出来的数原样填进来,只认字符串的话那次调用会被
|
||||||
|
# 判成参数不合法,白花一次调用去纠正一个与本层无关的形式问题。
|
||||||
|
"properties": {"answer": {"type": ["string", "number"]}},
|
||||||
|
"required": ["answer"],
|
||||||
|
"additionalProperties": False,
|
||||||
|
},
|
||||||
|
completes_run=True,
|
||||||
|
handler=_submit,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 事件出口
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class _RecordingEventSink:
|
||||||
|
"""把收到的事件记进列表。一次运行一个实例,不跨运行复用。"""
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self.events: list[Event] = []
|
||||||
|
|
||||||
|
async def emit(self, event: Event) -> None:
|
||||||
|
self.events.append(event)
|
||||||
|
|
||||||
|
def parameters(self) -> Mapping[str, str]:
|
||||||
|
return {"kind": "recording"}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 装配
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def model_client():
|
||||||
|
"""一个连着真实网关的模型客户端。
|
||||||
|
|
||||||
|
客户端持有连接池,用完必须 `aclose()`——不关的话每个用例漏一份连接池,而表现只是事件
|
||||||
|
循环关闭时的一串告警。
|
||||||
|
"""
|
||||||
|
client = GatewayClient.from_env()
|
||||||
|
try:
|
||||||
|
yield GatewayModelClient(client=client, settings=GatewaySettings.from_env())
|
||||||
|
finally:
|
||||||
|
await client.aclose()
|
||||||
|
|
||||||
|
|
||||||
|
def _text(role: Role, text: str) -> Message:
|
||||||
|
return Message(role=role, content=(TextBlock(text=text),))
|
||||||
|
|
||||||
|
|
||||||
|
def _definition(model_client, store: JsonlRunStore, sink: _RecordingEventSink) -> AgentDefinition:
|
||||||
|
return AgentDefinition(
|
||||||
|
model_client=model_client,
|
||||||
|
decision_parser=_JsonDecisionParser(),
|
||||||
|
store=store,
|
||||||
|
event_sink=sink,
|
||||||
|
synthetic_observations=SyntheticObservations(
|
||||||
|
action_rejected="这次工具调用没有执行:工具名或参数不合法。改过之后重新输出一个 JSON 对象。",
|
||||||
|
env_failed="环境出错了,这次工具调用没有产生结果。",
|
||||||
|
model_call_failed="上一次模型调用失败了。",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _request(run_id: str) -> RunRequest:
|
||||||
|
"""一次运行的装配。
|
||||||
|
|
||||||
|
`max_steps` 取 3:这次运行正常走完是两步(算一次、提交一次),留一步的余量给模型偶尔多说
|
||||||
|
一轮。**上限压得这么低是为了控制费用**——这一层每跑一次都在花钱,而它要证明的事(链路通不通)
|
||||||
|
两步就证明完了。
|
||||||
|
"""
|
||||||
|
registry = _registry()
|
||||||
|
return RunRequest(
|
||||||
|
run_id=run_id,
|
||||||
|
budget=Budget(
|
||||||
|
max_steps=3,
|
||||||
|
max_actions=3,
|
||||||
|
max_consecutive_parse_failures=2,
|
||||||
|
max_prompt_chars=20_000,
|
||||||
|
),
|
||||||
|
action_executor=registry.executor(),
|
||||||
|
tools=registry,
|
||||||
|
context=Context(
|
||||||
|
run_level=(_text(Role.SYSTEM, _PROTOCOL),),
|
||||||
|
goal_level=(_text(Role.USER, _GOAL),),
|
||||||
|
),
|
||||||
|
injections={},
|
||||||
|
model_binding={},
|
||||||
|
model_replay_policy=ReplayPolicy.NEVER,
|
||||||
|
observation_template=_OBSERVATION_TEMPLATE,
|
||||||
|
cancel_grace_seconds=5.0,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 用例一:完整闭环
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
#: 这次运行走完之后可以落在的两个停止原因。
|
||||||
|
#:
|
||||||
|
#: 两个都算正常终态,因为两者都意味着**循环自己走到了头**:`TASK_COMPLETED` 是模型调了带完成
|
||||||
|
#: 标记的工具,`AGENT_FINISHED` 是模型算完之后直接给了最终回答。只写前一个的话,模型选了后
|
||||||
|
#: 一条同样合法的路,这条测试就会红。预算耗尽、连续解析失败、模型调用失败都不在里面——那些
|
||||||
|
#: 是循环没走通。
|
||||||
|
_CLOSED_LOOP_STOP_REASONS = (StopReason.TASK_COMPLETED, StopReason.AGENT_FINISHED)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_a_real_run_goes_from_model_through_a_tool_and_back(model_client, tmp_path) -> None:
|
||||||
|
"""一次运行走通「模型 → 解释 → 工具执行 → 观察回填 → 再问模型 → 收尾」。"""
|
||||||
|
store = JsonlRunStore(directory=tmp_path)
|
||||||
|
sink = _RecordingEventSink()
|
||||||
|
request = _request("e2e-closed-loop")
|
||||||
|
|
||||||
|
result = await run(_definition(model_client, store, sink), request)
|
||||||
|
|
||||||
|
assert result.stop_reason in _CLOSED_LOOP_STOP_REASONS
|
||||||
|
# 只走一次模型调用不算闭环:那种运行证明的只是「请求发得出去」,证明不了观察回填之后模型
|
||||||
|
# 还能接着往下走。
|
||||||
|
assert len(result.steps) >= 2
|
||||||
|
assert any(step.action_status is ActionStatus.EXECUTED for step in result.steps)
|
||||||
|
for step in result.steps:
|
||||||
|
# 调用标识是「真的打出去过」的硬证据:它由网关那边生成,替身给不出来。
|
||||||
|
assert step.call_id, f"第 {step.step_idx} 步没有调用标识"
|
||||||
|
assert step.prompt_chars > 0, f"第 {step.step_idx} 步的提示词规模是 0"
|
||||||
|
assert step.step_wall_ms > 0, f"第 {step.step_idx} 步的墙钟是 0"
|
||||||
|
|
||||||
|
# 往返等价:日志读回来的步序列与返回值逐字段相等。不等的话,下游拿轨迹做的分析和拿返回值
|
||||||
|
# 做的分析会得出不同的结论,而两边都自称是这次运行。
|
||||||
|
log = await store.read_log(request.run_id)
|
||||||
|
assert tuple(entry.step for entry in log.steps) == result.steps
|
||||||
|
assert log.finished is not None
|
||||||
|
assert log.finished.result.stop_reason is result.stop_reason
|
||||||
|
|
||||||
|
assert len(sink.events) == len(result.steps)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 用例二:取消穿透
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class _EntryAnnouncingClient:
|
||||||
|
"""转发给真客户端,并在进入那次调用时打一个信号。
|
||||||
|
|
||||||
|
取消要落在**真实的 HTTP 请求途中**才验得到东西。靠 `sleep` 猜时机的话,慢一点就落在解释
|
||||||
|
或工具执行上、快一点就落在调用发出之前,而两种落空都表现成一条绿的测试。这个信号把时机
|
||||||
|
收成确定的:它一亮,下一个 await 就是那次真实请求。
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, inner) -> None:
|
||||||
|
self._inner = inner
|
||||||
|
self.entered = asyncio.Event()
|
||||||
|
|
||||||
|
async def call(self, call: ModelCall) -> ModelReply:
|
||||||
|
self.entered.set()
|
||||||
|
return await self._inner.call(call)
|
||||||
|
|
||||||
|
def parameters(self) -> Mapping[str, str]:
|
||||||
|
return self._inner.parameters()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_cancellation_passes_through_a_real_model_call(model_client, tmp_path) -> None:
|
||||||
|
"""取消能穿过真实的网络调用,并且留下一条以「已取消」收尾的结束记录。
|
||||||
|
|
||||||
|
结束记录是这条测试的另一半:没有它,恢复读到的是一次没有结束标记的运行,会被当成可以
|
||||||
|
续跑——而它其实是被人主动叫停的。
|
||||||
|
"""
|
||||||
|
store = JsonlRunStore(directory=tmp_path)
|
||||||
|
sink = _RecordingEventSink()
|
||||||
|
request = _request("e2e-cancelled")
|
||||||
|
client = _EntryAnnouncingClient(model_client)
|
||||||
|
|
||||||
|
task = asyncio.create_task(run(_definition(client, store, sink), request))
|
||||||
|
await asyncio.wait_for(client.entered.wait(), timeout=30)
|
||||||
|
# 信号亮起时那次请求还没被 await。让出一小会儿,取消就确实落在请求飞在网上的那段。
|
||||||
|
await asyncio.sleep(0.5)
|
||||||
|
task.cancel()
|
||||||
|
|
||||||
|
with pytest.raises(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
|
||||||
|
log = await store.read_log(request.run_id)
|
||||||
|
assert log.finished is not None, "取消之后没有结束记录,这次运行看起来还能续跑"
|
||||||
|
assert log.finished.result.stop_reason is StopReason.CANCELLED
|
||||||
Reference in New Issue
Block a user