iomgaa
1fac387e75
tests/ 不进 wheel，所以那套被 CLAUDE.md §0 称作「任何新适配器的准入标准」的用例，第一个 下游根本拿不到。**接法同时换掉**：pytest 的 conftest 只沿被收集文件的目录链查找，装在 site-packages 里的测试模块看不见下游的 conftest，原来那个「在自己的 conftest 里覆盖同名 fixture」的接法在发布之后走不通。改成继承契约基类，下游的子类定义在自己的目录链上。 **搬的过程中发现这套准入标准从来没被执行过。** test_model_client.py 有四条用例调用 records.model_call(...)，而工厂里根本没有这个方法——它没炸是因为那个 fixture 默认 skip。 五个接缝里只有存储那套被真跑过（15 条跳过里有 15 条是这四套）。 所以这个提交的另一半是让它真的跑起来。存储接两个实现（一份契约同时验多个实现，正是换接法 换来的）；动作执行接注册表分发器，外加一个有真实等待点的替身，否则那条取消用例的断言半边 永远走不到；模型调用接网关适配器，落在 integration，它连的是真网关；决策解释与事件出口各 接一个测试替身——替身住在 tests/ 里不进 wheel，下游拿不到，所以不违反「库不带默认实现」， 判据是下游拿不拿得到。 **一并清掉两类坏用例。** 五条函数体只有 docstring、一个断言都没有却报 PASSED 的假绿——一个 准入标准里出现假绿比出现跳过糟得多，下游看到全绿会以为验过了。以及一条端口从没承诺过的 长度断言（len(history_text) <= len(reply.content)）：压测的 AppWorld 场景为了迁就它，刻意 不补被复刻的实现真的会补的三个反引号，注释里写着「补一个字符就违约」。七条「这一层验不了」 统一成无条件 skip，理由字符串写全「承诺是什么／为什么验不了／你该在哪儿自己验」。 **发一个 pytest11 entry point，只为换回断言重写。** 契约模块不在下游的 python_files 里， 默认不被重写，于是一条契约失败时下游看到的是光秃秃的 AssertionError。不做的话没有任何东西 会报错，纯静默退化。实测过：editable 安装下 entry point 注册了但重写不生效（RECORD 里没有 包文件），要装真 wheel 才验得出来。
178 lines
6.6 KiB
Python
178 lines
6.6 KiB
Python
"""记录工厂造出来的东西必须合法：构造得出，而且能过一遍编解码往返。
|
||
|
||
**它守的是工厂造的记录合不合法，不是「契约用例引用的方法存不存在」。** 后者只有真的执行那条
|
||
用例才查得出——一条用例调了工厂上不存在的方法，工厂自己的单元测试怎么写都看不见它，因为那
|
||
条引用根本不在这个文件里。所以这份测试全绿不代表契约套件接上了实现；把五套契约都接到实现上
|
||
是另一件事，做在 `tests/contract/` 与 `tests/integration/`
|
||
（`research-wiki/design/0014-contract-suite-distribution.md` 决策六）。
|
||
|
||
往返用的是 `polyloop.serialization`，因为那是记录进日志的唯一通道：一条编不出来或者解回来
|
||
不等于自己的记录，在契约套件里表现成某个存储实现的用例红，而红的原因其实在工厂这一侧。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from polyloop.serialization import (
|
||
decode_intent,
|
||
decode_model_call_result,
|
||
decode_run_finished,
|
||
decode_run_result,
|
||
decode_run_started,
|
||
decode_step_completed,
|
||
decode_step_record,
|
||
encode,
|
||
)
|
||
from polyloop.testing import RecordFactory
|
||
from polyloop.types import ActionStatus, ReplayPolicy, StopReason
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
@pytest.fixture
|
||
def records() -> RecordFactory:
|
||
return RecordFactory()
|
||
|
||
|
||
def test_run_started_round_trips(records: RecordFactory) -> None:
|
||
record = records.run_started(run_id="r1", parameter_snapshot={"model": "m-1"})
|
||
|
||
assert decode_run_started(encode(record)) == record
|
||
|
||
|
||
def test_model_call_intent_round_trips(records: RecordFactory) -> None:
|
||
record = records.model_call_intent(run_id="r1", call_index=0, result_id="m0")
|
||
|
||
assert decode_intent(encode(record)) == record
|
||
|
||
|
||
def test_action_intent_round_trips_with_a_non_default_replay_policy(
|
||
records: RecordFactory,
|
||
) -> None:
|
||
"""重放策略跟着走一遍。它是恢复时「这一步要不要重跑」的输入，编码里丢了就会静默降级。"""
|
||
record = records.action_intent(
|
||
run_id="r1", call_index=1, result_id="a0", replay_policy=ReplayPolicy.SAFE
|
||
)
|
||
|
||
assert decode_intent(encode(record)) == record
|
||
assert record.replay_policy is ReplayPolicy.SAFE
|
||
|
||
|
||
def test_successful_model_call_result_round_trips(records: RecordFactory) -> None:
|
||
record = records.model_call_result(run_id="r1", result_id="m0", reply=records.reply())
|
||
|
||
assert decode_model_call_result(encode(record)) == record
|
||
|
||
|
||
def test_failed_model_call_result_round_trips(records: RecordFactory) -> None:
|
||
"""失败那一档单独走一遍：回复为空、失败说明有值，两个可空字段的组合和成功那档相反。"""
|
||
record = records.model_call_result(run_id="r1", result_id="m0", failure="连接超时")
|
||
|
||
decoded = decode_model_call_result(encode(record))
|
||
|
||
assert decoded == record
|
||
assert decoded.reply is None
|
||
assert decoded.failure == "连接超时"
|
||
|
||
|
||
def test_step_record_round_trips(records: RecordFactory) -> None:
|
||
record = records.step(step_idx=3)
|
||
|
||
assert decode_step_record(encode(record)) == record
|
||
|
||
|
||
def test_unparsed_step_record_round_trips(records: RecordFactory) -> None:
|
||
"""解析失败那一档：动作为空、解析说明有值。"""
|
||
record = records.step(parse_ok=False)
|
||
|
||
decoded = decode_step_record(encode(record))
|
||
|
||
assert decoded == record
|
||
assert decoded.action is None
|
||
assert decoded.parse_error is not None
|
||
|
||
|
||
def test_step_completed_round_trips(records: RecordFactory) -> None:
|
||
record = records.step_completed(
|
||
run_id="r1",
|
||
result_id="a0",
|
||
action_outcome=records.outcome(),
|
||
step=records.step(),
|
||
)
|
||
|
||
assert decode_step_completed(encode(record)) == record
|
||
|
||
|
||
def test_step_completed_without_an_action_round_trips(records: RecordFactory) -> None:
|
||
"""没有动作的那一步：结果标识与动作结果同时为空，这个组合存储要能原样存下来。"""
|
||
record = records.step_completed(
|
||
run_id="r1", result_id=None, action_outcome=None, step=records.step()
|
||
)
|
||
|
||
decoded = decode_step_completed(encode(record))
|
||
|
||
assert decoded == record
|
||
assert decoded.result_id is None
|
||
assert decoded.action_outcome is None
|
||
|
||
|
||
def test_outcome_carries_a_non_default_status(records: RecordFactory) -> None:
|
||
"""状态取值跟着记录走一遍。默认那档和显式传的那档在编码里长得一样，各验一次。"""
|
||
record = records.step_completed(
|
||
run_id="r1",
|
||
result_id="a0",
|
||
action_outcome=records.outcome(status=ActionStatus.ENV_ERROR),
|
||
step=records.step(),
|
||
)
|
||
|
||
decoded = decode_step_completed(encode(record))
|
||
|
||
assert decoded == record
|
||
assert decoded.action_outcome is not None
|
||
assert decoded.action_outcome.status is ActionStatus.ENV_ERROR
|
||
|
||
|
||
def test_run_result_round_trips(records: RecordFactory) -> None:
|
||
record = records.result(run_id="r1", stop_reason=StopReason.STEP_BUDGET)
|
||
|
||
assert decode_run_result(encode(record)) == record
|
||
|
||
|
||
def test_run_finished_round_trips(records: RecordFactory) -> None:
|
||
record = records.run_finished(run_id="r1", result=records.result(run_id="r1"))
|
||
|
||
assert decode_run_finished(encode(record)) == record
|
||
|
||
|
||
def test_model_call_defaults_to_one_user_message(records: RecordFactory) -> None:
|
||
"""一次模型调用不是记录，编解码不管它，但它的默认消息序列是契约用例的隐式输入。
|
||
|
||
默认给一条用户消息而不是空序列：一个真实的实现拿到空消息序列多半直接拒绝，于是那几条
|
||
用例验的就变成了它的入参校验，不是它的返回结构。
|
||
"""
|
||
call = records.model_call(call_index=0, result_id="m0")
|
||
|
||
assert len(call.messages) == 1
|
||
assert call.messages[0].content[0].text != ""
|
||
|
||
|
||
def test_action_with_a_tool_name_carries_a_tool_call(records: RecordFactory) -> None:
|
||
"""带工具名的动作要真的带上工具调用，按工具名分发的执行器靠它才走得进分发。"""
|
||
action = records.action(text="echo", tool_name="echo")
|
||
|
||
assert action.tool_call is not None
|
||
assert action.tool_call.name == "echo"
|
||
|
||
|
||
def test_action_without_a_tool_name_carries_no_tool_call(records: RecordFactory) -> None:
|
||
"""不带工具名的那种是代码执行型动作，工具调用必须为空，否则会被分发器当成工具调用收下。"""
|
||
assert records.action().tool_call is None
|
||
|
||
|
||
def test_event_carries_the_step_it_reports(records: RecordFactory) -> None:
|
||
"""事件不是记录，但它带着的那条步记录要和工厂造的其他步记录同形。"""
|
||
event = records.event(step_idx=2)
|
||
|
||
assert event.step is not None
|
||
assert event.step.step_idx == 2
|
||
assert decode_step_record(encode(event.step)) == event.step
|