262 lines
9.1 KiB
Python
262 lines
9.1 KiB
Python
"""慢更新动量生成（app/harness/momentum.py）单元测试。
|
|
|
|
覆盖:
|
|
- _categorize_pair 四分类 + 缺键 KeyError
|
|
- _format_comparison_pairs 分组排序 + 空列表
|
|
- run_slow_momentum 正常解析 + 解析失败保留 prev_guidance
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING, Any
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from app.harness.momentum import (
|
|
IMPROVED,
|
|
PERSISTENT_FAIL,
|
|
REGRESSED,
|
|
STABLE_SUCCESS,
|
|
_categorize_pair,
|
|
_format_comparison_pairs,
|
|
run_slow_momentum,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
|
|
# =========================================================================
|
|
# _categorize_pair
|
|
# =========================================================================
|
|
|
|
|
|
class TestCategorizePair:
|
|
"""_categorize_pair 四分类测试。"""
|
|
|
|
def test_categorize_pair_all_four(self) -> None:
|
|
"""四种 correct_prev/correct_curr 组合应返回对应类别常量。"""
|
|
assert _categorize_pair({"correct_prev": False, "correct_curr": True}) == IMPROVED
|
|
assert _categorize_pair({"correct_prev": True, "correct_curr": False}) == REGRESSED
|
|
assert _categorize_pair({"correct_prev": False, "correct_curr": False}) == PERSISTENT_FAIL
|
|
assert _categorize_pair({"correct_prev": True, "correct_curr": True}) == STABLE_SUCCESS
|
|
|
|
def test_categorize_pair_truthy_values(self) -> None:
|
|
"""非布尔真值（int / str）也能正确分类。"""
|
|
assert _categorize_pair({"correct_prev": 0, "correct_curr": 1}) == IMPROVED
|
|
assert _categorize_pair({"correct_prev": "yes", "correct_curr": ""}) == REGRESSED
|
|
|
|
def test_categorize_pair_missing_key(self) -> None:
|
|
"""缺 correct_prev 或 correct_curr 键时抛 KeyError。"""
|
|
with pytest.raises(KeyError):
|
|
_categorize_pair({"correct_prev": True})
|
|
with pytest.raises(KeyError):
|
|
_categorize_pair({"correct_curr": False})
|
|
with pytest.raises(KeyError):
|
|
_categorize_pair({})
|
|
|
|
|
|
# =========================================================================
|
|
# _format_comparison_pairs
|
|
# =========================================================================
|
|
|
|
|
|
class TestFormatComparisonPairs:
|
|
"""_format_comparison_pairs 分组排序测试。"""
|
|
|
|
def test_format_comparison_pairs_empty(self) -> None:
|
|
"""空列表返回占位说明。"""
|
|
result = _format_comparison_pairs([])
|
|
assert "无可用" in result
|
|
|
|
def test_format_comparison_pairs_order(self) -> None:
|
|
"""REGRESSED 标题应出现在 IMPROVED 标题之前（伤害信号优先）。"""
|
|
pairs = [
|
|
{
|
|
"question": "Q1",
|
|
"prev_prediction": "A",
|
|
"curr_prediction": "B",
|
|
"correct_prev": False,
|
|
"correct_curr": True,
|
|
},
|
|
{
|
|
"question": "Q2",
|
|
"prev_prediction": "C",
|
|
"curr_prediction": "D",
|
|
"correct_prev": True,
|
|
"correct_curr": False,
|
|
},
|
|
]
|
|
result = _format_comparison_pairs(pairs)
|
|
regressed_pos = result.index("回退")
|
|
improved_pos = result.index("改善")
|
|
assert regressed_pos < improved_pos, "REGRESSED 应排在 IMPROVED 之前"
|
|
|
|
def test_format_comparison_pairs_all_categories(self) -> None:
|
|
"""四种类别的 pair 都能被正确分组。"""
|
|
pairs = [
|
|
{"question": "Q1", "correct_prev": False, "correct_curr": True},
|
|
{"question": "Q2", "correct_prev": True, "correct_curr": False},
|
|
{"question": "Q3", "correct_prev": False, "correct_curr": False},
|
|
{"question": "Q4", "correct_prev": True, "correct_curr": True},
|
|
]
|
|
result = _format_comparison_pairs(pairs)
|
|
assert "固定样本总数：4" in result
|
|
# 每个类别都应标注 1 题
|
|
assert "1 题" in result
|
|
|
|
def test_format_comparison_pairs_missing_key(self) -> None:
|
|
"""pair 缺键时 KeyError 不被吞。"""
|
|
with pytest.raises(KeyError):
|
|
_format_comparison_pairs([{"question": "Q1"}])
|
|
|
|
|
|
# =========================================================================
|
|
# run_slow_momentum
|
|
# =========================================================================
|
|
|
|
|
|
def _make_mock_llm(response_content: str) -> Any:
|
|
"""构造一个返回指定 content 的 mock LLMProvider。"""
|
|
|
|
@dataclass(frozen=True)
|
|
class _FakeResponse:
|
|
content: str
|
|
thinking: str = ""
|
|
model: str = "mock"
|
|
provider: str = "mock"
|
|
prompt_tokens: int = 0
|
|
completion_tokens: int = 0
|
|
latency_ms: int = 0
|
|
ttft_ms: float | None = None
|
|
max_inter_token_ms: float | None = None
|
|
cache_hit: bool = False
|
|
call_id: str = "test-call-id"
|
|
|
|
mock_llm = AsyncMock()
|
|
mock_llm.chat.return_value = _FakeResponse(content=response_content)
|
|
return mock_llm
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slow_momentum_basic(tmp_path: Path) -> None:
|
|
"""正常解析时返回新的动量指导文本。"""
|
|
# 准备 prompt 文件
|
|
prompt_file = tmp_path / "slow_momentum.md"
|
|
prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
|
|
|
|
new_guidance_text = "新一轮的动量指导内容"
|
|
llm_response = json.dumps({"slow_update_content": new_guidance_text}, ensure_ascii=False)
|
|
mock_llm = _make_mock_llm(llm_response)
|
|
|
|
result = await run_slow_momentum(
|
|
llm=mock_llm,
|
|
diagnose_prompts_dir=tmp_path,
|
|
skill_content="当前 skill 正文",
|
|
prev_skill="上一版 skill 正文",
|
|
prev_guidance="旧的动量指导",
|
|
comparison_pairs=[
|
|
{
|
|
"question": "Q1",
|
|
"prev_prediction": "A",
|
|
"curr_prediction": "B",
|
|
"correct_prev": False,
|
|
"correct_curr": True,
|
|
}
|
|
],
|
|
)
|
|
assert result == new_guidance_text
|
|
mock_llm.chat.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slow_momentum_parse_failure(tmp_path: Path) -> None:
|
|
"""LLM 返回无法解析的内容时保留 prev_guidance。"""
|
|
prompt_file = tmp_path / "slow_momentum.md"
|
|
prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
|
|
|
|
mock_llm = _make_mock_llm("这不是 JSON，无法解析")
|
|
prev_guidance = "应该被保留的旧动量指导"
|
|
|
|
result = await run_slow_momentum(
|
|
llm=mock_llm,
|
|
diagnose_prompts_dir=tmp_path,
|
|
skill_content="当前 skill",
|
|
prev_skill="上一版 skill",
|
|
prev_guidance=prev_guidance,
|
|
comparison_pairs=[
|
|
{
|
|
"question": "Q1",
|
|
"prev_prediction": "A",
|
|
"curr_prediction": "B",
|
|
"correct_prev": True,
|
|
"correct_curr": True,
|
|
}
|
|
],
|
|
)
|
|
assert result == prev_guidance
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slow_momentum_missing_field(tmp_path: Path) -> None:
|
|
"""LLM 返回合法 JSON 但缺少 slow_update_content 字段时保留 prev_guidance。"""
|
|
prompt_file = tmp_path / "slow_momentum.md"
|
|
prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
|
|
|
|
llm_response = json.dumps({"other_field": "无关内容"}, ensure_ascii=False)
|
|
mock_llm = _make_mock_llm(llm_response)
|
|
prev_guidance = "应该被保留的旧动量指导"
|
|
|
|
result = await run_slow_momentum(
|
|
llm=mock_llm,
|
|
diagnose_prompts_dir=tmp_path,
|
|
skill_content="当前 skill",
|
|
prev_skill="上一版 skill",
|
|
prev_guidance=prev_guidance,
|
|
comparison_pairs=[],
|
|
)
|
|
assert result == prev_guidance
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slow_momentum_keyerror_not_swallowed(tmp_path: Path) -> None:
|
|
"""comparison_pairs 缺键时 KeyError 不被 ValueError 吞掉。"""
|
|
prompt_file = tmp_path / "slow_momentum.md"
|
|
prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
|
|
|
|
mock_llm = _make_mock_llm('{"slow_update_content": "ok"}')
|
|
|
|
with pytest.raises(KeyError):
|
|
await run_slow_momentum(
|
|
llm=mock_llm,
|
|
diagnose_prompts_dir=tmp_path,
|
|
skill_content="当前 skill",
|
|
prev_skill="上一版 skill",
|
|
prev_guidance="旧指导",
|
|
comparison_pairs=[{"question": "Q1"}], # 缺 correct_prev/correct_curr
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slow_momentum_fenced_json(tmp_path: Path) -> None:
|
|
"""LLM 返回 fenced code block 中的 JSON 也能正确解析。"""
|
|
prompt_file = tmp_path / "slow_momentum.md"
|
|
prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
|
|
|
|
new_guidance = "从 fenced block 中提取的指导"
|
|
llm_response = f'```json\n{{"slow_update_content": "{new_guidance}"}}\n```'
|
|
mock_llm = _make_mock_llm(llm_response)
|
|
|
|
result = await run_slow_momentum(
|
|
llm=mock_llm,
|
|
diagnose_prompts_dir=tmp_path,
|
|
skill_content="当前 skill",
|
|
prev_skill="上一版 skill",
|
|
prev_guidance="旧指导",
|
|
comparison_pairs=[],
|
|
)
|
|
assert result == new_guidance
|