Files
Video-Tree-TRM5/tests/unit/test_evolve.py
T

660 lines
22 KiB
Python

"""core/evolution/evolve.py 单元测试。
覆盖验证函数、编辑预算退火、resolve_skill_file、内部辅助函数、
受保护区构建、JSON 解析、rank_and_clip、格式化工具。
"""
from __future__ import annotations
import json
from unittest.mock import AsyncMock
import pytest
from core.evolution.evolve import (
_check_code_blocks,
_check_length,
_extract_section,
_format_case_samples,
_format_rejected_edits,
_format_spans,
_parse_frontmatter,
_parse_llm_json,
_select_top_edits,
_skill_protected_spans,
_strip_appendix_region,
_strip_momentum_region,
_strip_protected_regions,
_system_protected_spans,
_tool_protected_spans,
edit_budget_at,
rank_and_clip,
resolve_skill_file,
validate_skill,
validate_system,
validate_tool,
)
from core.evolution.patch import (
APPENDIX_END,
APPENDIX_START,
MOMENTUM_END,
MOMENTUM_START,
)
from core.evolution.types import RejectedEdit
# =========================================================================
# A. 内部辅助函数
# =========================================================================
class TestParseFrontmatter:
"""_parse_frontmatter 测试。"""
def test_valid_frontmatter(self) -> None:
text = "---\nname: test\ndescription: d\n---\nbody"
result = _parse_frontmatter(text)
assert result == {"name": "test", "description": "d"}
def test_no_frontmatter(self) -> None:
assert _parse_frontmatter("no frontmatter here") is None
def test_invalid_yaml(self) -> None:
text = "---\n: : invalid\n---\nbody"
assert _parse_frontmatter(text) is None
def test_empty_frontmatter(self) -> None:
text = "---\n\n---\nbody"
result = _parse_frontmatter(text)
assert result is None # yaml.safe_load("") returns None
class TestStripAppendixRegion:
"""_strip_appendix_region 测试。"""
def test_no_markers(self) -> None:
text = "hello world"
assert _strip_appendix_region(text) == text
def test_with_markers(self) -> None:
text = f"before\n{APPENDIX_START}\nappendix stuff\n{APPENDIX_END}\nafter"
result = _strip_appendix_region(text)
assert "appendix stuff" not in result
assert "before" in result
assert "after" in result
def test_only_start_marker(self) -> None:
text = f"before\n{APPENDIX_START}\nno end marker"
assert _strip_appendix_region(text) == text
def test_only_end_marker(self) -> None:
text = f"before\n{APPENDIX_END}\nno start marker"
assert _strip_appendix_region(text) == text
class TestStripMomentumRegion:
"""_strip_momentum_region 测试。"""
def test_no_markers(self) -> None:
text = "hello world"
assert _strip_momentum_region(text) == text
def test_with_markers(self) -> None:
text = f"before\n{MOMENTUM_START}\nmomentum stuff\n{MOMENTUM_END}\nafter"
result = _strip_momentum_region(text)
assert "momentum stuff" not in result
assert "before" in result
assert "after" in result
def test_damaged_markers_raise(self) -> None:
text = f"before\n{MOMENTUM_START}\nno end marker"
with pytest.raises(ValueError, match="momentum marker 损坏"):
_strip_momentum_region(text)
class TestStripProtectedRegions:
"""_strip_protected_regions 测试(先 appendix 后 momentum)。"""
def test_both_regions(self) -> None:
text = (
f"body\n"
f"{APPENDIX_START}\nappendix\n{APPENDIX_END}\n"
f"{MOMENTUM_START}\nmomentum\n{MOMENTUM_END}\n"
f"tail"
)
result = _strip_protected_regions(text)
assert "appendix" not in result
assert "momentum" not in result
assert "body" in result
assert "tail" in result
class TestCheckLength:
"""_check_length 测试。"""
def test_normal_ratio(self) -> None:
orig = "x" * 100
evol = "x" * 120
assert _check_length(orig, evol) == []
def test_too_long(self) -> None:
orig = "x" * 100
evol = "x" * 300
errors = _check_length(orig, evol)
assert len(errors) == 1
assert "超限" in errors[0]
def test_too_short(self) -> None:
orig = "x" * 100
evol = "x" * 10
errors = _check_length(orig, evol)
assert len(errors) == 1
assert "不足" in errors[0]
def test_orig_empty(self) -> None:
assert _check_length("", "something") == []
class TestCheckCodeBlocks:
"""_check_code_blocks 测试。"""
def test_even_count(self) -> None:
text = "```python\ncode\n```"
assert _check_code_blocks(text) == []
def test_odd_count(self) -> None:
text = "```python\ncode"
errors = _check_code_blocks(text)
assert len(errors) == 1
assert "未闭合" in errors[0]
def test_no_blocks(self) -> None:
assert _check_code_blocks("no code blocks") == []
class TestExtractSection:
"""_extract_section 测试。"""
def test_found(self) -> None:
text = "intro\n## 能力边界\nfrozen content\n## other\nmore"
result = _extract_section(text, "能力边界")
assert result is not None
assert "frozen content" in result
assert "## 能力边界" in result
def test_not_found(self) -> None:
text = "intro\n## other\nmore"
assert _extract_section(text, "不存在") is None
def test_last_section(self) -> None:
text = "intro\n## 最后段\ncontent at end"
result = _extract_section(text, "最后段")
assert result is not None
assert "content at end" in result
# =========================================================================
# B. 受保护区构建
# =========================================================================
class TestSkillProtectedSpans:
"""_skill_protected_spans 测试。"""
def test_with_frontmatter(self) -> None:
text = "---\nname: test\n---\nbody"
spans = _skill_protected_spans(text)
assert any("---" in s for s in spans)
def test_with_appendix(self) -> None:
text = f"body\n{APPENDIX_START}\nnotes\n{APPENDIX_END}"
spans = _skill_protected_spans(text)
assert any(APPENDIX_START in s for s in spans)
def test_with_momentum(self) -> None:
text = f"body\n{MOMENTUM_START}\nmomentum\n{MOMENTUM_END}"
spans = _skill_protected_spans(text)
assert any(MOMENTUM_START in s for s in spans)
def test_empty(self) -> None:
assert _skill_protected_spans("plain text") == []
class TestSystemProtectedSpans:
"""_system_protected_spans 测试。"""
def test_frozen_sections(self) -> None:
text = "intro\n## 能力边界\ncontent1\n## 输出格式\ncontent2\n## 视频树结构\ncontent3\n## other\nmore"
spans = _system_protected_spans(text)
assert len(spans) == 3
def test_with_appendix(self) -> None:
text = f"body\n{APPENDIX_START}\nnotes\n{APPENDIX_END}"
spans = _system_protected_spans(text)
assert len(spans) == 1
class TestToolProtectedSpans:
"""_tool_protected_spans 测试。"""
def test_output_format_section(self) -> None:
text = "intro\n## 输出格式\nformat\n## other\nmore"
spans = _tool_protected_spans(text)
assert any("输出格式" in s for s in spans)
def test_no_output_format(self) -> None:
text = "intro\n## other\nmore"
spans = _tool_protected_spans(text)
assert len(spans) == 0
# =========================================================================
# C. 验证函数
# =========================================================================
class TestValidateSkill:
"""validate_skill 测试。"""
def test_identical_passes(self) -> None:
content = "---\nname: test\ndescription: d\ntask_type: t\n---\nbody"
assert validate_skill(content, content).passed
def test_changed_frontmatter_fails(self) -> None:
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody"
evol = "---\nname: b\ndescription: d\ntask_type: t\n---\nbody"
result = validate_skill(orig, evol)
assert not result.passed
assert any("name" in e for e in result.errors)
def test_length_ratio_too_short_fails(self) -> None:
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\n" + "x" * 1000
evol = "---\nname: a\ndescription: d\ntask_type: t\n---\nshort"
result = validate_skill(orig, evol)
assert not result.passed
assert any("不足" in e for e in result.errors)
def test_length_ratio_too_long_fails(self) -> None:
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nshort"
evol = "---\nname: a\ndescription: d\ntask_type: t\n---\n" + "x" * 1000
result = validate_skill(orig, evol)
assert not result.passed
assert any("超限" in e for e in result.errors)
def test_unclosed_code_block_fails(self) -> None:
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody"
evol = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody\n```"
result = validate_skill(orig, evol)
assert not result.passed
assert any("未闭合" in e for e in result.errors)
def test_missing_original_frontmatter(self) -> None:
result = validate_skill("no frontmatter", "no frontmatter")
assert not result.passed
assert any("原文缺少" in e for e in result.errors)
def test_missing_evolved_frontmatter(self) -> None:
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody"
result = validate_skill(orig, "no frontmatter body")
assert not result.passed
class TestValidateSystem:
"""validate_system 测试。"""
def test_identical_passes(self) -> None:
content = "intro\n## 能力边界\nfrozen\n## 输出格式\nfrozen2\n## other\nbody"
assert validate_system(content, content).passed
def test_changed_frozen_section_fails(self) -> None:
orig = "intro\n## 能力边界\noriginal\n## other\nbody"
evol = "intro\n## 能力边界\nchanged\n## other\nbody"
result = validate_system(orig, evol)
assert not result.passed
assert any("能力边界" in e for e in result.errors)
def test_missing_frozen_section_fails(self) -> None:
orig = "intro\n## 能力边界\nfrozen\n## other\nbody"
evol = "intro\n## other\nbody that is similar length content padding"
result = validate_system(orig, evol)
assert not result.passed
assert any("缺失" in e for e in result.errors)
def test_no_frozen_sections_passes(self) -> None:
content = "intro\n## other section\nbody content"
assert validate_system(content, content).passed
def test_code_block_check(self) -> None:
content = "## 能力边界\nfrozen\n## other\nbody\n```unclosed"
result = validate_system(content, content)
assert not result.passed
class TestValidateTool:
"""validate_tool 测试。"""
def test_identical_passes(self) -> None:
extract = "## 输出格式\nfixed\n## other\nbody"
verify = "## 输出格式\nfixed2\n## other\nbody2"
assert validate_tool(extract, extract, verify, verify).passed
def test_no_code_block_check(self) -> None:
"""validate_tool 不检查代码块闭合(与 skill/system 不同)。"""
extract = "## 输出格式\nfixed\n```\nunclosed"
assert validate_tool(extract, extract, "v", "v").passed
def test_changed_output_format_fails(self) -> None:
orig = "## 输出格式\noriginal format\n## other\nbody"
evol = "## 输出格式\nchanged format\n## other\nbody"
result = validate_tool(orig, evol, "v", "v")
assert not result.passed
assert any("输出格式" in e for e in result.errors)
def test_verify_output_format_checked_too(self) -> None:
extract = "## 输出格式\nfixed\n## other\nbody"
orig_verify = "## 输出格式\nfixed_v\n## other\nbody_v"
evol_verify = "## 输出格式\nchanged_v\n## other\nbody_v"
result = validate_tool(extract, extract, orig_verify, evol_verify)
assert not result.passed
# =========================================================================
# D. 纯数学
# =========================================================================
class TestEditBudget:
"""edit_budget_at 测试。"""
def test_start_at_zero(self) -> None:
assert edit_budget_at(0, 100, 5, 2) == 5
def test_end_at_total(self) -> None:
assert edit_budget_at(100, 100, 5, 2) == 2
def test_total_steps_one(self) -> None:
assert edit_budget_at(0, 1, 5, 2) == 5
def test_start_less_than_end_asserts(self) -> None:
with pytest.raises(AssertionError):
edit_budget_at(0, 100, 2, 5)
def test_mid_step(self) -> None:
result = edit_budget_at(50, 100, 5, 2)
assert 2 <= result <= 5
def test_beyond_total_clamped(self) -> None:
"""global_step 超过 total_steps 时被钳住在 end。"""
assert edit_budget_at(200, 100, 5, 2) == 2
def test_equal_start_end(self) -> None:
assert edit_budget_at(50, 100, 3, 3) == 3
def test_total_steps_zero(self) -> None:
"""total_steps <= 1 直接返回 start。"""
assert edit_budget_at(0, 0, 5, 2) == 5
# =========================================================================
# E. JSON 解析
# =========================================================================
class TestParseLlmJson:
"""_parse_llm_json 测试。"""
def test_plain_json(self) -> None:
raw = '{"key": "value"}'
assert _parse_llm_json(raw) == {"key": "value"}
def test_fenced_json(self) -> None:
raw = 'text before\n```json\n{"key": "value"}\n```\ntext after'
assert _parse_llm_json(raw) == {"key": "value"}
def test_non_dict_returns_none(self) -> None:
raw = "[1, 2, 3]"
assert _parse_llm_json(raw) is None
def test_invalid_json_returns_none(self) -> None:
raw = "not json at all"
assert _parse_llm_json(raw) is None
def test_empty_string(self) -> None:
assert _parse_llm_json("") is None
# =========================================================================
# F. rank_and_clip
# =========================================================================
class TestSelectTopEdits:
"""_select_top_edits 测试。"""
def test_valid_indices(self) -> None:
edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}]
result = _select_top_edits([2, 0], edits, 2)
assert result == [{"op": "c"}, {"op": "a"}]
def test_dedup(self) -> None:
edits = [{"op": "a"}, {"op": "b"}]
result = _select_top_edits([0, 0, 1], edits, 3)
assert len(result) == 2
def test_out_of_bounds_skipped(self) -> None:
edits = [{"op": "a"}, {"op": "b"}]
result = _select_top_edits([5, 0, -1], edits, 3)
assert result == [{"op": "a"}]
def test_bool_excluded(self) -> None:
"""type(True) is int 返回 True 但 type(idx) is int 排除 bool。"""
edits = [{"op": "a"}, {"op": "b"}]
result = _select_top_edits([True, False, 0], edits, 3)
assert result == [{"op": "a"}]
def test_max_edits_limit(self) -> None:
edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}]
result = _select_top_edits([0, 1, 2], edits, 2)
assert len(result) == 2
class TestRankAndClip:
"""rank_and_clip 测试。"""
@pytest.mark.asyncio
async def test_within_budget_passthrough(self) -> None:
"""edits <= max_edits 时原样返回。"""
llm = AsyncMock()
edits = [{"op": "a"}, {"op": "b"}]
result, info = await rank_and_clip(llm, "content", edits, 5, "test")
assert result == edits
assert info["triggered"] is False
llm.chat.assert_not_called()
@pytest.mark.asyncio
async def test_llm_rank_success(self) -> None:
"""LLM 成功返回索引时按优先级裁剪。"""
llm = AsyncMock()
response = AsyncMock()
response.content = json.dumps({"selected_indices": [2, 0]})
llm.chat.return_value = response
edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}]
result, info = await rank_and_clip(llm, "content", edits, 2, "test")
assert len(result) == 2
assert result[0] == {"op": "c"}
assert info["triggered"] is True
@pytest.mark.asyncio
async def test_llm_failure_fallback(self) -> None:
"""LLM 失败时退化为按原序取前 max_edits 条。"""
llm = AsyncMock()
llm.chat.side_effect = Exception("LLM down")
edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}]
result, info = await rank_and_clip(llm, "content", edits, 2, "test")
assert len(result) == 2
assert result == [{"op": "a"}, {"op": "b"}]
assert info["triggered"] is True
# =========================================================================
# G. resolve_skill_file
# =========================================================================
class TestResolveSkillFile:
"""resolve_skill_file 测试。"""
def test_direct_match(self) -> None:
class FakeStore:
def list_skill_files(self) -> list[str]:
return ["action-reasoning.md", "default-strategy.md"]
def read_skill(self, f: str) -> str:
return ""
assert resolve_skill_file(FakeStore(), "Action Reasoning") == "action-reasoning.md"
def test_fallback_to_default(self) -> None:
class FakeStore:
def list_skill_files(self) -> list[str]:
return ["default-strategy.md"]
def read_skill(self, f: str) -> str:
return ""
assert resolve_skill_file(FakeStore(), "Unknown Type") == "default-strategy.md"
def test_case_insensitive(self) -> None:
class FakeStore:
def list_skill_files(self) -> list[str]:
return ["temporal-reasoning.md"]
def read_skill(self, f: str) -> str:
return ""
assert resolve_skill_file(FakeStore(), "Temporal Reasoning") == "temporal-reasoning.md"
# =========================================================================
# H. 格式化辅助
# =========================================================================
class TestFormatCaseSamples:
"""_format_case_samples 测试。"""
def test_basic_format(self) -> None:
cases = [
{
"question_id": "q1",
"question": "What?",
"options": ["A", "B"],
"answer": "A",
"prediction": "B",
"error_type": "wrong",
"selection_reason": "test",
"trace": [],
}
]
result = _format_case_samples(cases)
assert "q1" in result
assert "What?" in result
def test_trace_truncation(self) -> None:
cases = [
{
"question_id": "q1",
"question": "Q",
"options": [],
"answer": "A",
"prediction": "B",
"error_type": "e",
"selection_reason": "s",
"trace": [
{
"step": 1,
"tool_name": "t",
"tool_args": {},
"tool_output": "x" * 600,
}
],
}
]
result = _format_case_samples(cases)
assert "..." in result
class TestFormatSpans:
"""_format_spans 测试。"""
def test_basic_format(self) -> None:
spans = [
{
"step": 1,
"tool_name": "extract",
"tool_args": {"query": "test"},
"tool_output": "result",
"extraction_completeness": 0.9,
"hallucination_rate": 0.1,
}
]
result = _format_spans(spans)
assert "extract" in result
assert "0.9" in result
def test_output_truncation(self) -> None:
spans = [
{
"step": 1,
"tool_name": "t",
"tool_args": {},
"tool_output": "x" * 600,
"extraction_completeness": 0.5,
"hallucination_rate": 0.0,
}
]
result = _format_spans(spans)
assert "..." in result
class TestFormatRejectedEdits:
"""_format_rejected_edits 测试。"""
def test_with_gate_evidence(self) -> None:
edits = [
RejectedEdit(
target_file="skill.md",
target_type="skill",
change_summary="changed X",
delta=-0.05,
source_version="v2",
epoch=3,
gate_w=5,
gate_l=8,
gate_e_value=0.42,
gate_delta_shrunk=-0.123,
)
]
result = _format_rejected_edits(edits)
assert "W=5" in result
assert "L=8" in result
assert "E=0.42" in result
assert "δ̂=-0.123" in result
def test_without_gate_evidence(self) -> None:
edits = [
RejectedEdit(
target_file="skill.md",
target_type="skill",
change_summary="changed X",
delta=-0.05,
source_version="v2",
epoch=3,
)
]
result = _format_rejected_edits(edits)
assert "skill.md" in result
assert "changed X" in result
assert "W=" not in result