"""统一完成判定在 transport 两条路径上的落点(#30/#31,设计 §4.2-§4.4、§5.4)。 假 SSE 按 magpie/glm 实测形态二次构造: 末块 finish_reason + usage 后干净关闭、 不发 [DONE];以及只有思考的 length 截断、null/未知字段等方言。 """ import asyncio import json import httpx import pytest from loguru import logger from polygateway.errors import AllSourcesExhausted, ResultInvalidError, TransientError from polygateway.transports.openai_compat import OpenAICompatTransport from tests.unit.test_openai_compat import _complete, _source, _sse_stream, _transport_for _USAGE = {"prompt_tokens": 11, "completion_tokens": 7} _TOOL = { "index": 0, "id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": '{"q": 1}'}, } def _f(delta=None, finish=None, usage=None, choices=True): """单 choice 帧;`choices=False` 构造 usage-only 帧。""" body = {"id": "up-1"} if choices: choice = {"index": 0, "delta": delta or {}} if finish is not None: choice["finish_reason"] = finish body["choices"] = [choice] else: body["choices"] = [] if usage is not None: body["usage"] = usage return f"data: {json.dumps(body)}\n\n" def _assistant(**delta): return _f({"role": "assistant", **delta}) async def _run(frames, *, done, mode="text", **source_kw): transport = _transport_for(lambda _: _sse_stream(*frames, done=done)) try: return await _complete(transport, _source(**source_kw), response_mode=mode) finally: await transport.aclose() async def _run_once(body, *, mode="text"): transport = _transport_for(lambda _: httpx.Response(200, json=body)) try: return await _complete(transport, _source(), stream=False, response_mode=mode) finally: await transport.aclose() # —— #31: 缺 [DONE] 但 finish + 其后实测 usage 完整 —— @pytest.mark.parametrize("mode", ["text", "openai"]) @pytest.mark.parametrize("finish", ["stop", "length"]) async def test_eof_after_finish_with_usage_succeeds(mode, finish): """验收 1: 末块 finish + usage 后干净关闭 → 成功、measured、eof_after_finish。""" frames = [_assistant(content="hel"), _f({"content": "lo"}, finish=finish, usage=_USAGE)] result = await _run(frames, done=False, mode=mode) assert result.content == "hello" and result.stream_end == "eof_after_finish" assert result.usage_source == "measured" and result.finish_reason == finish assert (result.prompt_tokens, result.completion_tokens) == (11, 7) if mode == "openai": assert result.completion.finish_reason == finish async def test_eof_after_finish_usage_only_frame_and_tools(): """验收 2: tool_calls 结束、usage 在其后的 usage-only 帧 → 工具完整交回。""" frames = [ _assistant(content=None, tool_calls=[_TOOL]), _f({}, finish="tool_calls"), _f(choices=False, usage=_USAGE), ] result = await _run(frames, done=False, mode="openai") tool = result.completion.message["tool_calls"][0] assert tool["function"] == {"name": "lookup", "arguments": '{"q": 1}'} assert result.stream_end == "eof_after_finish" and result.usage_source == "measured" @pytest.mark.parametrize( "frames", [ [_assistant(content="partial"), _f(choices=False, usage=_USAGE)], # 验收 3: 无 finish [_assistant(content="partial"), _f({}, finish="stop")], # 验收 4: 无 usage [_assistant(content="a", **{}), _f({}, usage=_USAGE), _f({}, finish="stop")], # usage 早 [_assistant(content="a"), _f({}, finish="stop", usage={})], # usage 不可用 ], ) async def test_incomplete_eof_still_fails(frames): """截断不放宽: 保真仍不打捞,文本 retry 档仍抛。""" with pytest.raises(TransientError, match="missing_done"): await _run(frames, done=False, mode="openai", missing_done="salvage") with pytest.raises(TransientError, match="missing_done"): await _run(frames, done=False) async def test_text_salvage_marks_stream_end(): """文本 salvage 打捞 → stream_end=salvaged,usage 降为 estimated。""" frames = [_assistant(content="partial"), _f(choices=False, usage=_USAGE)] result = await _run(frames, done=False, missing_done="salvage") assert result.stream_end == "salvaged" and result.usage_source == "estimated" async def test_read_error_mid_stream_fails(): """验收 5: 读取中途 RemoteProtocolError → 失败,不当干净 EOF。""" class Broken(httpx.AsyncByteStream): async def __aiter__(self): yield _f({"role": "assistant", "content": "a"}, finish="stop", usage=_USAGE).encode() raise httpx.RemoteProtocolError("peer closed") async def aclose(self): return None transport = _transport_for(lambda _: httpx.Response(200, stream=Broken())) try: with pytest.raises(TransientError, match="网络错误"): await _complete(transport, _source(), response_mode="openai") finally: await transport.aclose() @pytest.mark.parametrize("mode", ["text", "openai"]) async def test_done_stream_marks_done(mode): """验收 6: 带 [DONE] 的正常流行为不变,stream_end=done。""" result = await _run([_assistant(content="ok"), _f({}, "stop", _USAGE)], done=True, mode=mode) assert result.stream_end == "done" and result.finish_reason == "stop" @pytest.mark.parametrize("done", [True, False]) @pytest.mark.parametrize("mode", ["text", "openai"]) async def test_upstream_error_finish_is_transient(done, mode): """GLM network_error: 上游自报失败,有正文也不交付。""" frames = [_assistant(content="half"), _f({}, finish="network_error", usage=_USAGE)] with pytest.raises(TransientError, match="upstream_finish_error"): await _run(frames, done=done, mode=mode) # —— #30 与结局表(设计 §4.3)—— _THINK = [_assistant(reasoning_content="long thought")] @pytest.mark.parametrize("done", [True, False]) async def test_openai_reasoning_only_length_delivered(done): """#30 验收: 只有思考 + length → 交回,content 空,思考完整。""" result = await _run([*_THINK, _f({}, "length", _USAGE)], done=done, mode="openai") message = result.completion.message assert message.get("content") in (None, "") and message["reasoning_content"] == "long thought" assert result.finish_reason == "length" and result.thinking == "long thought" @pytest.mark.parametrize( ("frames", "finish", "openai_ok", "text_error"), [ (_THINK, "length", True, (ResultInvalidError, "length_without_content")), ([_assistant()], "length", True, (ResultInvalidError, "length_without_content")), (_THINK, "content_filter", True, (ResultInvalidError, "filtered_without_content")), ([_assistant()], "sensitive", True, (ResultInvalidError, "filtered_without_content")), (_THINK, "stop", True, (TransientError, "empty_completion")), ([_assistant(refusal="no")], "stop", True, (TransientError, "empty_completion")), ([_assistant()], "stop", False, (TransientError, "empty_completion")), ], ) async def test_outcome_rows_stream(frames, finish, openai_ok, text_error): """行 3/4/6/7/8: 两模式的分叉。""" full = [*frames, _f({}, finish, _USAGE)] if openai_ok: assert (await _run(full, done=True, mode="openai")).finish_reason == finish else: with pytest.raises(TransientError, match="empty_completion"): await _run(full, done=True, mode="openai") error, code = text_error with pytest.raises(error, match=code) as caught: await _run(full, done=True) assert caught.type is error # ResultInvalid 不得被 Transient 子类掩盖 @pytest.mark.parametrize("mode", ["text", "openai"]) async def test_reasoning_only_without_finish_is_transient(mode): """行 9: 只靠 DONE 收尾、无 finish → 无法区分抖动与结局,维持瞬时。""" with pytest.raises(TransientError, match="empty_completion"): await _run([*_THINK, _f(choices=False, usage=_USAGE)], done=True, mode=mode) async def test_openai_whitespace_content_stop_delivered(): """正文只有空白 + stop: 保真交回(#30 一-2)。""" result = await _run( [_assistant(content=" "), _f({}, "stop", _USAGE)], done=True, mode="openai" ) assert result.completion.message["content"] == " " async def test_truncated_tool_is_result_invalid(): """行 2: length 截断的半截工具(缺 name)→ ResultInvalid tool_truncated。""" broken = {"index": 0, "id": "c", "type": "function", "function": {"arguments": '{"q'}} frames = [_assistant(tool_calls=[broken]), _f({}, "length", _USAGE)] with pytest.raises(ResultInvalidError, match="tool_truncated"): await _run(frames, done=True, mode="openai") async def test_content_cannot_mask_broken_tool(): """行 2: 正文非空 + 坏工具 + stop → Transient tool_incomplete。""" broken = {"index": 0, "id": "c", "type": "function", "function": {"arguments": "{}"}} frames = [_assistant(content="text", tool_calls=[broken]), _f({}, "stop", _USAGE)] with pytest.raises(TransientError, match="tool_incomplete"): await _run(frames, done=True, mode="openai") async def test_null_refusal_and_tool_calls_stream(): """#30 验收二: delta 带 refusal:null 与 tool_calls:null → 正常交回。""" frames = [_assistant(content="ok", refusal=None, tool_calls=None), _f({}, "stop", _USAGE)] result = await _run(frames, done=True, mode="openai") assert result.completion.message["content"] == "ok" # —— 非流式镜像 —— def _body(message, finish): choice = {"index": 0, "message": {"role": "assistant", **message}} if finish is not None: choice["finish_reason"] = finish return {"id": "up", "choices": [choice], "usage": _USAGE} async def test_nonstream_null_tool_calls_and_refusal(): """非流式 message 带 tool_calls:null + refusal:null → 成功。""" body = _body({"content": "hi", "tool_calls": None, "refusal": None}, "stop") result = await _run_once(body, mode="openai") assert result.completion.message["tool_calls"] is None and result.stream_end is None @pytest.mark.parametrize( ("message", "finish", "openai_expect", "text_expect"), [ ( {"reasoning_content": "t"}, "length", None, (ResultInvalidError, "length_without_content"), ), ({"content": ""}, "content_filter", None, (ResultInvalidError, "filtered_without_content")), ({"reasoning_content": "t"}, "stop", None, (TransientError, "empty_completion")), ({"reasoning_content": "t"}, None, (TransientError, "empty_completion"), None), ({"content": "a"}, "network_error", (TransientError, "upstream_finish_error"), None), ({"content": "a"}, "length", None, None), ], ) async def test_outcome_rows_nonstream(message, finish, openai_expect, text_expect): """非流式跳过流完整性,直接以 body finish 走结局表。""" body = _body(message, finish) for mode, expect in (("openai", openai_expect), ("text", text_expect or openai_expect)): if expect is None: result = await _run_once(body, mode=mode) assert result.finish_reason == finish and result.stream_end is None else: with pytest.raises(expect[0], match=expect[1]) as caught: await _run_once(body, mode=mode) assert caught.type is expect[0] async def test_nonstream_truncated_tool_is_result_invalid(): """非流式先判定后构造快照: 缺 name + length → tool_truncated 而非构造期 Transient。""" tool = {"id": "c", "type": "function", "function": {"arguments": "{"}} with pytest.raises(ResultInvalidError, match="tool_truncated"): await _run_once(_body({"tool_calls": [tool]}, "length"), mode="openai") # —— reasoning 别名(设计 §4.4)—— async def test_reasoning_alias_projects_thinking(): """只有 reasoning 字段: 文本 thinking 取别名;非空 reasoning_content 优先。""" alias = await _run( [_assistant(reasoning="r1"), _f({"content": "a"}, "stop", _USAGE)], done=True ) assert alias.thinking == "r1" both = [ _assistant(reasoning="r1", reasoning_content="rc"), _f({"content": "a"}, "stop", _USAGE), ] assert (await _run(both, done=True)).thinking == "rc" empty_rc = [_assistant(reasoning="r1", reasoning_content=""), _f({"content": "a"}, "stop")] assert (await _run(empty_rc, done=True)).thinking == "r1" nonstream = await _run_once(_body({"content": "a", "reasoning": "nr"}, "stop")) assert nonstream.thinking == "nr" async def test_reasoning_alias_keeps_ttft_alive(): """只有 reasoning 的思考期刷新活性,不被 TTFT 看门狗误杀。""" class Slow(httpx.AsyncByteStream): async def __aiter__(self): yield _assistant().encode() for _ in range(5): # 0.5s 只有 reasoning,远超 ttft,但每帧间隔 < inter_token yield _f({"reasoning": "t"}).encode() await asyncio.sleep(0.1) yield _f({"content": "answer"}, "stop", _USAGE).encode() yield b"data: [DONE]\n\n" async def aclose(self): return None transport = _transport_for(lambda _: httpx.Response(200, stream=Slow())) try: source = _source(ttft_timeout_s=0.3, inter_token_timeout_s=0.25) result = await _complete(transport, source) assert result.content == "answer" and result.thinking == "ttttt" finally: await transport.aclose() # —— 有界告警节流(设计 §5.4)—— def _sink(): records = [] handle = logger.add(lambda m: records.append(m.record["message"]), level="WARNING") return records, handle async def test_conflict_and_unknown_finish_warn_once(): """同源同路径冲突、同一未知 finish 值各只告警一次,不回显字段值。""" frames = [ _assistant(content="a", ext=1), _f({"ext": 2}), _f({}, "end_turn", _USAGE), ] transport = _transport_for(lambda _: _sse_stream(*frames)) records, handle = _sink() try: for _ in range(2): result = await _complete(transport, _source(), response_mode="openai") assert "ext" not in result.completion.message finally: logger.remove(handle) await transport.aclose() conflict = [r for r in records if "delta.ext" in r] unknown = [r for r in records if "end_turn" in r] assert len(conflict) == 1 and len(unknown) == 1 and "不缓存" in unknown[0] async def test_warning_set_is_bounded(): """节流集合饱和后只记一次饱和告警,不再增长。""" transport = OpenAICompatTransport() transport._warned_fields.update((f"s{i}", "k") for i in range(1024)) records, handle = _sink() try: transport._warn_once("qwen_1", "field:x", "a") transport._warn_once("qwen_1", "field:y", "b") finally: logger.remove(handle) await transport.aclose() assert len(transport._warned_fields) == 1024 assert sum("饱和" in r for r in records) == 1 # —— 治理链: 确定性结局不重试、不计熔断 —— async def test_text_length_without_content_not_retried_nor_breaker_failure(): """文本 length 无正文 → ResultInvalid 直抛,attempts=1,熔断不记失败。""" from tests.unit.test_client import _client calls = [] def handler(_): calls.append(1) return _sse_stream(*_THINK, _f({}, "length", _USAGE)) client = _client(handler=handler) failures = [] original = client._breaker_backend.record_failure async def spy(*args, **kwargs): failures.append(args) return await original(*args, **kwargs) client._breaker_backend.record_failure = spy try: with pytest.raises(ResultInvalidError, match="length_without_content"): await client.chat([{"role": "user", "content": "hi"}]) finally: await client._transport.aclose() assert len(calls) == 1 and failures == [] async def test_openai_reasoning_length_single_attempt_via_client(): """#30 验收经治理链: attempts=1;#31 形态经治理链不再重试到耗尽。""" from tests.unit.test_client import _client calls = [] def handler(_): calls.append(1) return _sse_stream(*_THINK, _f({}, "length", _USAGE), done=False) client = _client(handler=handler) try: response = await client.chat([{"role": "user", "content": "hi"}], response_mode="openai") except AllSourcesExhausted: # pragma: no cover - 失败即回归 pytest.fail("reasoning-only length 不应重试到耗尽") finally: await client._transport.aclose() assert len(calls) == 1 and response.call_stats.attempts == 1 assert response.stream_end == "eof_after_finish" and response.finish_reason == "length" async def test_eof_after_finish_stop_single_attempt_via_client(): """#31 验收 1 经治理链: stop + usage 无 DONE,attempts=1、不重试到耗尽。""" from tests.unit.test_client import _client calls = [] def handler(_): calls.append(1) return _sse_stream(_assistant(content="ok"), _f({}, "stop", _USAGE), done=False) client = _client(handler=handler) try: response = await client.chat([{"role": "user", "content": "hi"}]) finally: await client._transport.aclose() assert len(calls) == 1 and response.call_stats.attempts == 1 assert response.stream_end == "eof_after_finish" and response.usage_source == "measured" async def test_text_salvage_with_length_partial_content_delivered(): """文本 salvage + length + 部分正文(finish 后无 usage): 交付、salvaged。""" frames = [_assistant(content="part"), _f({}, finish="length")] result = await _run(frames, done=False, missing_done="salvage") assert result.content == "part" and result.stream_end == "salvaged" assert result.finish_reason == "length" async def test_null_arguments_tool_is_result_invalid_stream(): """审查 I2: 无参函数 arguments 全程 null → ResultInvalid,不重试。""" tool = {"index": 0, "id": "c", "type": "function", "function": {"name": "f", "arguments": None}} frames = [_assistant(tool_calls=[tool]), _f({}, "tool_calls", _USAGE)] with pytest.raises(ResultInvalidError, match="tool_arguments_null"): await _run(frames, done=True, mode="openai")