iomgaa
20a4a9ae47
The M3 evidence sat at 08-02 for twenty-three days while nobody could tell whether it still held. A declaration that goes stale in silence is the failure this issue is really about, so the library now compares what it declared against what it just observed and says so when the two part ways. Judgement is separated from logging: reconcile_thinking returns the warning text, so tests assert on the text instead of parsing logs. Two cases that look alike are kept apart — a model whose capability is registered gets a drift warning quoting its evidence, an unregistered one is never told the table said anything, because it never did. False x UNKNOWN stays silent on purpose. UNKNOWN cannot falsify anything, and warning on it would fire on every disabled call M3 makes over the plain endpoint. A warning that always fires is not a warning.
280 lines
11 KiB
Python
280 lines
11 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
|
||
)
|
||
|
||
|
||
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)。"""
|
||
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_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
|
||
)
|