iomgaa
1307a02b92
Three of them were the same shape as the bug this branch exists to fix: something goes wrong, the library swallows it, and the caller is left with a number that means the opposite of what happened. The throttle key had no source in it. Five sources on one model is the normal case here, so the first one to break would warn once and silence the other four for the life of the process, and the message never said which gateway to look at. An unknown verdict in a cached entry threw away the whole response. The rehydrator tolerates unknown fields but not unknown values of a known field, so two library versions sharing a Redis would each invalidate the other's entries: halved hit rate, and the only log line says the cache rebuild failed. A purely observational field should not be able to void a response whose content is intact. Normalising for telemetry now degrades instead of raising, both for a bare string and for a value outside the domain. Either one used to reach the same except and cost the whole row, which is exactly how 1.3.0 lost nineteen calls without anyone noticing.
314 lines
13 KiB
Python
314 lines
13 KiB
Python
"""推理裁定与对账的行为测试(issue #16/#17 设计 §4-§5)。
|
||
|
||
判据来自 2026-08-25 实测(findings): MiniMax-M3 在开启档流式路径下返回 185 字符
|
||
推理正文却不上报 `completion_tokens_details`,而 qwen/deepseek 两者都报。库因此
|
||
不能把任何单一信号当权威——本组用例逐条钉死"哪个信号该赢"。
|
||
"""
|
||
|
||
import pytest
|
||
from loguru import logger
|
||
|
||
from polygateway.providers import get_provider
|
||
from polygateway.thinking import (
|
||
DEFAULT_CAPABILITIES,
|
||
ThinkingCapability,
|
||
get_capability,
|
||
observe_thinking,
|
||
reconcile_thinking,
|
||
register_capability,
|
||
resolve_thinking,
|
||
)
|
||
from polygateway.types import ThinkingObservation
|
||
|
||
|
||
def _warnings():
|
||
"""捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。"""
|
||
messages: list[str] = []
|
||
sink_id = logger.add(messages.append, level="WARNING")
|
||
return messages, sink_id
|
||
|
||
|
||
class TestObserveThinking:
|
||
"""三态裁定: 证据硬度决定优先级,无信号一律 UNKNOWN。"""
|
||
|
||
def test_reasoning_text_alone_proves_it_happened(self):
|
||
"""推理正文是事实本身: 上游不报 token 数也照样成立(M3 流式实测形态)。"""
|
||
assert (
|
||
observe_thinking(thinking="先解方程 x+y=35", reasoning_tokens=None)
|
||
is ThinkingObservation.OBSERVED
|
||
)
|
||
|
||
def test_blank_text_is_not_evidence(self):
|
||
"""纯空白正文不算证据: 网关响应是外部输入,truthy 判据会把空格计成推理(P5)。"""
|
||
assert (
|
||
observe_thinking(thinking=" \n\t ", reasoning_tokens=None)
|
||
is ThinkingObservation.UNKNOWN
|
||
)
|
||
|
||
def test_positive_token_count_proves_it_happened(self):
|
||
"""无正文但上游报了推理用量(qwen 非流式形态)。"""
|
||
assert observe_thinking(thinking="", reasoning_tokens=205) is ThinkingObservation.OBSERVED
|
||
|
||
def test_zero_token_count_is_positive_evidence_of_absence(self):
|
||
"""`0` 是"上报了且为零",与"没上报"语义不同,故是 ABSENT 而非 UNKNOWN。"""
|
||
assert observe_thinking(thinking="", reasoning_tokens=0) is ThinkingObservation.ABSENT
|
||
|
||
def test_no_signal_at_all_stays_unknown(self):
|
||
"""M3 非流式开启档的真实形态: 推理已计费却既无正文也无 token 数。
|
||
|
||
判成 ABSENT 就是伪装成"没推理"——正是 issue #16/#17 的病根。
|
||
"""
|
||
assert observe_thinking(thinking="", reasoning_tokens=None) is ThinkingObservation.UNKNOWN
|
||
|
||
def test_text_outranks_a_zero_count(self):
|
||
"""转述与事实冲突时事实赢: 正文在,`reasoning_tokens=0` 不能翻案。"""
|
||
assert (
|
||
observe_thinking(thinking="想了想", reasoning_tokens=0) is ThinkingObservation.OBSERVED
|
||
)
|
||
|
||
@pytest.mark.parametrize("negative", [-1, -205])
|
||
def test_negative_token_count_is_not_evidence_of_absence(self, negative):
|
||
"""负数是坏数据,不是"上游明确上报未推理"这个最强的正面结论。
|
||
|
||
当前 transport 已在边界把负数归 `None`,所以这条走不通;但本函数的
|
||
docstring 自称"外部输入校验后使用",第二个 transport 直接填该值时,
|
||
`> 0 else ABSENT` 会给出一个方向相反的强结论。函数自身必须闭合(P5)。
|
||
"""
|
||
assert observe_thinking(thinking="", reasoning_tokens=negative) is (
|
||
ThinkingObservation.UNKNOWN
|
||
)
|
||
|
||
|
||
class TestThinkingObservationEnum:
|
||
def test_values_are_stable_strings(self):
|
||
"""取值进遥测落库,改名即历史数据断层。"""
|
||
assert ThinkingObservation.OBSERVED == "observed"
|
||
assert ThinkingObservation.ABSENT == "absent"
|
||
assert ThinkingObservation.UNKNOWN == "unknown"
|
||
|
||
def test_enum_lives_in_the_innermost_layer(self):
|
||
"""枚举必须定义在 `types.py`(最内层)。
|
||
|
||
它是 `LLMResponse` 的字段类型;定义在决策层 `thinking.py` 会让 `types.py`
|
||
反向 import 决策模块,违反 P7 依赖铁律(import-linter 契约执法)。
|
||
"""
|
||
assert ThinkingObservation.__module__ == "polygateway.types"
|
||
|
||
|
||
@pytest.mark.parametrize("bogus", ["", "OBSERVED", "yes", "none"])
|
||
def test_unknown_strings_are_rejected(bogus):
|
||
"""非法值必须抛 ValueError: 缓存回放与遥测归一化都靠它识别域外取值(设计 §6)。
|
||
|
||
两处接住这个 ValueError 后**降级而非作废**(缓存复活内容 + 记 UNKNOWN、遥测
|
||
照常落行),但降级的前提是构造器真的会拒绝——它一旦放行,域外取值就会一路
|
||
进到 `LLMResponse` 与遥测列里。
|
||
"""
|
||
with pytest.raises(ValueError):
|
||
ThinkingObservation(bogus)
|
||
|
||
|
||
class TestThinkingCapability:
|
||
"""issue #5: 能力按 model 登记——同一 provider 内部代际差异是决定性的。"""
|
||
|
||
def test_registered_models_carry_evidence(self):
|
||
"""登记必须附实测证据: 表会过期,没有出处就无从判断该不该信。"""
|
||
for model in ("MiniMax-M3", "MiniMax-M2.7", "MiniMax-M2.5"):
|
||
cap = get_capability(model)
|
||
assert cap is not None and cap.evidence.strip()
|
||
|
||
def test_m3_can_disable_but_m2x_cannot(self):
|
||
assert get_capability("MiniMax-M3").can_disable is True
|
||
assert get_capability("MiniMax-M2.7").can_disable is False
|
||
assert get_capability("MiniMax-M2.5").can_disable is False
|
||
|
||
def test_unregistered_model_is_unknown(self):
|
||
assert get_capability("some-brand-new-model") is None
|
||
|
||
def test_register_capability_is_pure(self):
|
||
table = register_capability("x-1", ThinkingCapability(True, "实测"))
|
||
assert get_capability("x-1", table=table) is not None
|
||
assert get_capability("x-1") is None # 默认表未被污染
|
||
|
||
def test_default_capabilities_mapping_is_read_only(self):
|
||
with pytest.raises(TypeError):
|
||
DEFAULT_CAPABILITIES["hack"] = None # type: ignore[index]
|
||
|
||
|
||
class TestResolveThinking:
|
||
"""五条判定规则(顺序即语义);设计 §5 真值表。"""
|
||
|
||
def test_rule1_none_injects_nothing(self):
|
||
got = resolve_thinking(get_provider("minimax"), None, None, model="MiniMax-M3")
|
||
assert got == {}
|
||
|
||
@pytest.mark.parametrize("enable", [True, False])
|
||
def test_rule2_unknown_shape_raises_and_points_the_way(self, enable):
|
||
with pytest.raises(ValueError, match="register_provider") as exc:
|
||
resolve_thinking(get_provider("openai"), None, enable, model="kimi-k3")
|
||
assert "extra_body" in str(exc.value)
|
||
|
||
def test_rule3_unregistered_model_warns_but_passes(self):
|
||
messages, sink_id = _warnings()
|
||
try:
|
||
got = resolve_thinking(get_provider("minimax"), None, False, model="MiniMax-M9")
|
||
finally:
|
||
logger.remove(sink_id)
|
||
assert got == {"reasoning_effort": "none"}
|
||
assert any("MiniMax-M9" in m for m in messages)
|
||
|
||
def test_rule4_cannot_disable_raises_with_the_model_name(self):
|
||
cap = get_capability("MiniMax-M2.7")
|
||
with pytest.raises(ValueError, match="MiniMax-M2.7"):
|
||
resolve_thinking(get_provider("minimax"), cap, False, model="MiniMax-M2.7")
|
||
|
||
def test_rule4_only_blocks_the_off_direction(self):
|
||
"""关不掉 ≠ 开不了: M2.x 默认就在推理,开的方向不该被拦。"""
|
||
cap = get_capability("MiniMax-M2.7")
|
||
got = resolve_thinking(get_provider("minimax"), cap, True, model="MiniMax-M2.7")
|
||
assert got == {"reasoning_effort": "medium"}
|
||
|
||
def test_rule5_normal_path(self):
|
||
cap = get_capability("MiniMax-M3")
|
||
assert resolve_thinking(get_provider("minimax"), cap, False, model="MiniMax-M3") == {
|
||
"reasoning_effort": "none"
|
||
}
|
||
|
||
def test_unknown_shape_beats_capability_check(self):
|
||
"""第 2 步先于第 4 步: 形态未知时无从注入,能力如何无关紧要。"""
|
||
cap = ThinkingCapability(can_disable=False, evidence="构造")
|
||
with pytest.raises(ValueError, match="register_provider"):
|
||
resolve_thinking(get_provider("openai"), cap, False, model="whatever")
|
||
|
||
|
||
class TestReconcileThinking:
|
||
"""声明 × 观测对账(设计 §5): 矛盾出文案,不表态出 None。
|
||
|
||
文案本身是被断言对象——判定与日志分离正是为此: 告警内容可直接比对,不必
|
||
去解析日志格式。
|
||
"""
|
||
|
||
_CAP = ThinkingCapability(
|
||
can_disable=True, evidence="2026-08-02 实测 reasoning_effort=none 可关闭"
|
||
)
|
||
|
||
def test_off_but_observed_with_a_registered_capability_blames_the_table(self):
|
||
"""已登记却实测推理了 = 能力表漂移: 必须附 evidence 与更新指路。"""
|
||
msg = reconcile_thinking(
|
||
enable_thinking=False,
|
||
observation=ThinkingObservation.OBSERVED,
|
||
capability=self._CAP,
|
||
model="MiniMax-M3",
|
||
)
|
||
assert msg is not None
|
||
assert "MiniMax-M3" in msg
|
||
assert "2026-08-02 实测 reasoning_effort=none 可关闭" in msg
|
||
assert "register_capability" in msg
|
||
|
||
def test_off_but_observed_unregistered_never_claims_a_table_entry(self):
|
||
"""未登记模型没有"能力表声称"这回事——说它就是撒谎。"""
|
||
msg = reconcile_thinking(
|
||
enable_thinking=False,
|
||
observation=ThinkingObservation.OBSERVED,
|
||
capability=None,
|
||
model="MiniMax-M9",
|
||
)
|
||
assert msg is not None
|
||
assert "MiniMax-M9" in msg
|
||
assert "能力表" not in msg
|
||
assert "register_capability" in msg
|
||
|
||
def test_registered_and_unregistered_wordings_differ(self):
|
||
registered = reconcile_thinking(
|
||
enable_thinking=False,
|
||
observation=ThinkingObservation.OBSERVED,
|
||
capability=self._CAP,
|
||
model="MiniMax-M3",
|
||
)
|
||
unregistered = reconcile_thinking(
|
||
enable_thinking=False,
|
||
observation=ThinkingObservation.OBSERVED,
|
||
capability=None,
|
||
model="MiniMax-M3",
|
||
)
|
||
assert registered != unregistered
|
||
|
||
@pytest.mark.parametrize("capability", [None, _CAP])
|
||
def test_on_but_absent_is_a_contradiction(self, capability):
|
||
"""上游明确上报未推理: 这是唯一的正面证伪,与能力表登记与否无关。"""
|
||
msg = reconcile_thinking(
|
||
enable_thinking=True,
|
||
observation=ThinkingObservation.ABSENT,
|
||
capability=capability,
|
||
model="qwen3.7-plus",
|
||
)
|
||
assert msg is not None
|
||
assert "qwen3.7-plus" in msg
|
||
|
||
@pytest.mark.parametrize("capability", [None, _CAP])
|
||
def test_on_but_unknown_admits_it_cannot_confirm(self, capability):
|
||
"""issue #17 的诚实版本: 明说"我注入了,但我看不见结果"。"""
|
||
msg = reconcile_thinking(
|
||
enable_thinking=True,
|
||
observation=ThinkingObservation.UNKNOWN,
|
||
capability=capability,
|
||
model="MiniMax-M3",
|
||
)
|
||
assert msg is not None
|
||
assert "MiniMax-M3" in msg
|
||
|
||
def test_off_and_absent_stays_silent(self):
|
||
"""要求关闭 + 上游明确上报未推理 = 要求被满足,没有可报的矛盾。
|
||
|
||
这一格与 `test_off_and_unknown_stays_silent` 的沉默理由**不同**: 那里是
|
||
"没有证伪力",这里是"正面证实要求已满足"。两者都必须沉默,漏测哪一格,
|
||
把 Phase 2 的判据写成 `is ABSENT` 之类的反向条件都不会被抓住。
|
||
"""
|
||
assert (
|
||
reconcile_thinking(
|
||
enable_thinking=False,
|
||
observation=ThinkingObservation.ABSENT,
|
||
capability=self._CAP,
|
||
model="qwen3.7-plus",
|
||
)
|
||
is None
|
||
)
|
||
|
||
def test_off_and_unknown_stays_silent(self):
|
||
"""UNKNOWN 没有证伪力: 拿它报警等于每次关闭调用都喊(M3 关闭档恒落此档)。"""
|
||
assert (
|
||
reconcile_thinking(
|
||
enable_thinking=False,
|
||
observation=ThinkingObservation.UNKNOWN,
|
||
capability=self._CAP,
|
||
model="MiniMax-M3",
|
||
)
|
||
is None
|
||
)
|
||
|
||
@pytest.mark.parametrize(
|
||
"observation",
|
||
[ThinkingObservation.OBSERVED, ThinkingObservation.ABSENT, ThinkingObservation.UNKNOWN],
|
||
)
|
||
def test_no_request_no_grievance(self, observation):
|
||
"""调用方不表态,就无从谈"违背"。"""
|
||
assert (
|
||
reconcile_thinking(
|
||
enable_thinking=None,
|
||
observation=observation,
|
||
capability=self._CAP,
|
||
model="MiniMax-M3",
|
||
)
|
||
is None
|
||
)
|
||
|
||
def test_on_and_observed_is_exactly_what_was_asked_for(self):
|
||
assert (
|
||
reconcile_thinking(
|
||
enable_thinking=True,
|
||
observation=ThinkingObservation.OBSERVED,
|
||
capability=self._CAP,
|
||
model="MiniMax-M3",
|
||
)
|
||
is None
|
||
)
|