451 lines
19 KiB
Python
451 lines
19 KiB
Python
"""patch.py 补丁引擎单元测试。
|
||
|
||
覆盖四大区域：
|
||
1. TestRegionBounds — appendix/momentum 边界定位与损坏态检测
|
||
2. TestAppendix — 追加/提取/替换 appendix 区
|
||
3. TestMomentum — 替换/提取 momentum 区
|
||
4. TestApplyPatch — apply_patch_with_report 的 4 种 op + 冻结区 + 异常
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from core.evolution.patch import (
|
||
APPENDIX_END,
|
||
APPENDIX_MAX_CHARS,
|
||
APPENDIX_START,
|
||
MOMENTUM_END,
|
||
MOMENTUM_HEADING,
|
||
MOMENTUM_MAX_CHARS,
|
||
MOMENTUM_START,
|
||
append_to_appendix,
|
||
appendix_region_bounds,
|
||
apply_patch_with_report,
|
||
extract_appendix_notes,
|
||
momentum_inner,
|
||
momentum_region_bounds,
|
||
replace_appendix_notes,
|
||
replace_momentum,
|
||
)
|
||
|
||
# ── TestRegionBounds ──────────────────────────────────────────────
|
||
|
||
|
||
class TestRegionBounds:
|
||
"""appendix_region_bounds / momentum_region_bounds 边界与损坏态检测。"""
|
||
|
||
# -- appendix --
|
||
|
||
def test_appendix_both_absent_returns_none(self) -> None:
|
||
assert appendix_region_bounds("no markers here") is None
|
||
|
||
def test_appendix_normal_pair(self) -> None:
|
||
text = f"head\n{APPENDIX_START}\nnotes\n{APPENDIX_END}\ntail"
|
||
start, end = appendix_region_bounds(text) # type: ignore[misc]
|
||
assert text[start:end].startswith(APPENDIX_START)
|
||
assert text[start:end].endswith(APPENDIX_END)
|
||
|
||
def test_appendix_only_start_raises(self) -> None:
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
appendix_region_bounds(f"head\n{APPENDIX_START}\nno end")
|
||
|
||
def test_appendix_only_end_raises(self) -> None:
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
appendix_region_bounds(f"head\n{APPENDIX_END}\nno start")
|
||
|
||
def test_appendix_repeated_start_raises(self) -> None:
|
||
text = f"{APPENDIX_START}\n{APPENDIX_START}\n{APPENDIX_END}"
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
appendix_region_bounds(text)
|
||
|
||
def test_appendix_reversed_raises(self) -> None:
|
||
text = f"{APPENDIX_END}\nbody\n{APPENDIX_START}"
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
appendix_region_bounds(text)
|
||
|
||
# -- momentum --
|
||
|
||
def test_momentum_both_absent_returns_none(self) -> None:
|
||
assert momentum_region_bounds("no markers here") is None
|
||
|
||
def test_momentum_normal_pair(self) -> None:
|
||
text = f"head\n{MOMENTUM_START}\nguidance\n{MOMENTUM_END}\ntail"
|
||
start, end = momentum_region_bounds(text) # type: ignore[misc]
|
||
assert text[start:end].startswith(MOMENTUM_START)
|
||
assert text[start:end].endswith(MOMENTUM_END)
|
||
|
||
def test_momentum_only_start_raises(self) -> None:
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
momentum_region_bounds(f"head\n{MOMENTUM_START}\nno end")
|
||
|
||
def test_momentum_only_end_raises(self) -> None:
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
momentum_region_bounds(f"head\n{MOMENTUM_END}\nno start")
|
||
|
||
def test_momentum_repeated_end_raises(self) -> None:
|
||
text = f"{MOMENTUM_START}\n{MOMENTUM_END}\n{MOMENTUM_END}"
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
momentum_region_bounds(text)
|
||
|
||
def test_momentum_reversed_raises(self) -> None:
|
||
text = f"{MOMENTUM_END}\nbody\n{MOMENTUM_START}"
|
||
with pytest.raises(ValueError, match="不配对"):
|
||
momentum_region_bounds(text)
|
||
|
||
|
||
# ── TestAppendix ──────────────────────────────────────────────────
|
||
|
||
|
||
class TestAppendix:
|
||
"""append_to_appendix / extract_appendix_notes / replace_appendix_notes。"""
|
||
|
||
def test_append_creates_region(self) -> None:
|
||
out = append_to_appendix("# Skill\nbody", ["reminder A"])
|
||
assert APPENDIX_START in out
|
||
assert APPENDIX_END in out
|
||
assert "- reminder A" in out
|
||
|
||
def test_append_accumulates(self) -> None:
|
||
step1 = append_to_appendix("# Skill\nbody", ["note 1"])
|
||
step2 = append_to_appendix(step1, ["note 2"])
|
||
assert "- note 1" in step2
|
||
assert "- note 2" in step2
|
||
|
||
def test_append_empty_notes_returns_unchanged(self) -> None:
|
||
original = "# Skill\nbody"
|
||
assert append_to_appendix(original, []) is original
|
||
|
||
def test_append_all_whitespace_notes_returns_unchanged(self) -> None:
|
||
original = "# Skill\nbody"
|
||
assert append_to_appendix(original, [" ", "\t", ""]) == original
|
||
|
||
def test_extract_notes_roundtrip(self) -> None:
|
||
content = append_to_appendix("# Skill\nbody", ["aaa", "bbb"])
|
||
notes = extract_appendix_notes(content)
|
||
assert notes == ["aaa", "bbb"]
|
||
|
||
def test_extract_notes_no_region(self) -> None:
|
||
assert extract_appendix_notes("no region") == []
|
||
|
||
def test_replace_notes_overwrites(self) -> None:
|
||
content = append_to_appendix("# Skill\nbody", ["old"])
|
||
replaced = replace_appendix_notes(content, ["new1", "new2"])
|
||
notes = extract_appendix_notes(replaced)
|
||
assert notes == ["new1", "new2"]
|
||
assert "old" not in replaced
|
||
|
||
def test_replace_notes_empty_deletes_region(self) -> None:
|
||
content = append_to_appendix("# Skill\nbody", ["old"])
|
||
replaced = replace_appendix_notes(content, [])
|
||
assert APPENDIX_START not in replaced
|
||
assert APPENDIX_END not in replaced
|
||
|
||
def test_replace_notes_no_region_creates(self) -> None:
|
||
replaced = replace_appendix_notes("# Skill\nbody", ["fresh"])
|
||
assert "- fresh" in replaced
|
||
assert APPENDIX_START in replaced
|
||
|
||
def test_replace_notes_no_region_empty_notes_unchanged(self) -> None:
|
||
original = "# Skill\nbody"
|
||
assert replace_appendix_notes(original, []) == original
|
||
|
||
def test_append_warns_on_exceeding_max_chars(self, caplog: pytest.LogCaptureFixture) -> None:
|
||
"""appendix 区超长时 loguru warning，不截断。"""
|
||
long_note = "x" * (APPENDIX_MAX_CHARS + 100)
|
||
with caplog.at_level("WARNING"):
|
||
out = append_to_appendix("body", [long_note])
|
||
assert long_note in out # 不截断
|
||
|
||
|
||
# ── TestMomentum ──────────────────────────────────────────────────
|
||
|
||
|
||
class TestMomentum:
|
||
"""replace_momentum / momentum_inner。"""
|
||
|
||
def test_replace_creates_region(self) -> None:
|
||
out = replace_momentum("# Skill\nbody", "focus on X")
|
||
assert MOMENTUM_START in out
|
||
assert MOMENTUM_END in out
|
||
assert "focus on X" in out
|
||
|
||
def test_replace_overwrites(self) -> None:
|
||
step1 = replace_momentum("# Skill\nbody", "old guidance")
|
||
step2 = replace_momentum(step1, "new guidance")
|
||
assert "new guidance" in step2
|
||
assert "old guidance" not in step2
|
||
|
||
def test_replace_empty_guidance_clears(self) -> None:
|
||
"""空 guidance 合法：清空旧动量、保留标题和 marker。"""
|
||
step1 = replace_momentum("# Skill\nbody", "old guidance")
|
||
step2 = replace_momentum(step1, "")
|
||
assert MOMENTUM_START in step2
|
||
assert MOMENTUM_END in step2
|
||
assert "old guidance" not in step2
|
||
assert MOMENTUM_HEADING in step2
|
||
|
||
def test_replace_rejects_marker_in_guidance(self) -> None:
|
||
with pytest.raises(ValueError, match="marker"):
|
||
replace_momentum("body", f"bad {MOMENTUM_START} injection")
|
||
|
||
def test_replace_rejects_end_marker_in_guidance(self) -> None:
|
||
with pytest.raises(ValueError, match="marker"):
|
||
replace_momentum("body", f"bad {MOMENTUM_END} injection")
|
||
|
||
def test_momentum_inner_returns_guidance(self) -> None:
|
||
content = replace_momentum("# Skill\nbody", "focus on X")
|
||
inner = momentum_inner(content)
|
||
assert inner == "focus on X"
|
||
|
||
def test_momentum_inner_no_region(self) -> None:
|
||
assert momentum_inner("no region") == ""
|
||
|
||
def test_momentum_inner_strips_heading(self) -> None:
|
||
content = replace_momentum("body", "some guidance")
|
||
inner = momentum_inner(content)
|
||
assert MOMENTUM_HEADING not in inner
|
||
assert inner == "some guidance"
|
||
|
||
def test_replace_warns_on_exceeding_max_chars(self, caplog: pytest.LogCaptureFixture) -> None:
|
||
"""momentum 区超长时 loguru warning，不截断。"""
|
||
long_guidance = "y" * (MOMENTUM_MAX_CHARS + 100)
|
||
with caplog.at_level("WARNING"):
|
||
out = replace_momentum("body", long_guidance)
|
||
assert long_guidance in out # 不截断
|
||
|
||
def test_coexists_with_appendix(self) -> None:
|
||
"""momentum 与 appendix 独立共存。"""
|
||
content = append_to_appendix("# Skill\nbody", ["note A"])
|
||
content = replace_momentum(content, "focus on X")
|
||
assert APPENDIX_START in content
|
||
assert APPENDIX_END in content
|
||
assert MOMENTUM_START in content
|
||
assert MOMENTUM_END in content
|
||
notes = extract_appendix_notes(content)
|
||
assert notes == ["note A"]
|
||
inner = momentum_inner(content)
|
||
assert inner == "focus on X"
|
||
|
||
|
||
# ── TestApplyPatch ────────────────────────────────────────────────
|
||
|
||
|
||
class TestApplyPatch:
|
||
"""apply_patch_with_report 的 4 种 op + 冻结区 + 异常处理。"""
|
||
|
||
def test_append(self) -> None:
|
||
content = "# Title\n\nbody text"
|
||
edits = [{"op": "append", "target": "", "content": "new section"}]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
assert "new section" in out
|
||
assert reports[0]["status"] == "applied_append"
|
||
assert reports[0]["index"] == 1
|
||
|
||
def test_insert_after_success(self) -> None:
|
||
content = "# Title\n\nanchor line\n\nrest"
|
||
edits = [{"op": "insert_after", "target": "anchor line", "content": "inserted"}]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
assert "inserted" in out
|
||
assert reports[0]["status"] == "applied_insert_after"
|
||
|
||
def test_insert_after_missing_target_fallback(self) -> None:
|
||
content = "# Title\n\nbody"
|
||
edits = [{"op": "insert_after", "target": "nonexistent", "content": "payload"}]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
assert "payload" in out
|
||
assert reports[0]["status"] == "applied_insert_after_fallback"
|
||
|
||
def test_insert_after_protected_skip(self) -> None:
|
||
content = "# Title\n\nFROZEN BLOCK\n\nrest"
|
||
edits = [{"op": "insert_after", "target": "FROZEN BLOCK", "content": "nope"}]
|
||
out, reports = apply_patch_with_report(content, edits, protected_spans=["FROZEN BLOCK"])
|
||
assert out == content
|
||
assert reports[0]["status"] == "skipped_protected"
|
||
|
||
def test_replace(self) -> None:
|
||
content = "# Title\n\nold text\n\nrest"
|
||
edits = [{"op": "replace", "target": "old text", "content": "new text"}]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
assert "new text" in out
|
||
assert "old text" not in out
|
||
assert reports[0]["status"] == "applied_replace"
|
||
|
||
def test_replace_protected_skip(self) -> None:
|
||
content = "# Title\n\nprotected\n\nrest"
|
||
edits = [{"op": "replace", "target": "protected", "content": "nope"}]
|
||
out, reports = apply_patch_with_report(content, edits, protected_spans=["protected"])
|
||
assert "protected" in out
|
||
assert reports[0]["status"] == "skipped_protected"
|
||
|
||
def test_delete(self) -> None:
|
||
content = "# Title\n\nremove me\n\nrest"
|
||
edits = [{"op": "delete", "target": "remove me", "content": ""}]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
assert "remove me" not in out
|
||
assert reports[0]["status"] == "applied_delete"
|
||
|
||
def test_unknown_op(self) -> None:
|
||
edits = [{"op": "magic", "target": "x", "content": "y"}]
|
||
out, reports = apply_patch_with_report("body", edits)
|
||
assert reports[0]["status"] == "skipped_unknown_op"
|
||
|
||
def test_non_dict_edit(self) -> None:
|
||
edits = ["not a dict"] # type: ignore[list-item]
|
||
out, reports = apply_patch_with_report("body", edits)
|
||
assert reports[0]["status"] == "error"
|
||
assert "非 dict" in reports[0].get("error", "")
|
||
|
||
def test_missing_target_for_replace(self) -> None:
|
||
edits = [{"op": "replace", "target": "", "content": "payload"}]
|
||
out, reports = apply_patch_with_report("body", edits)
|
||
assert reports[0]["status"] == "skipped_missing_target"
|
||
|
||
def test_missing_target_for_delete(self) -> None:
|
||
edits = [{"op": "delete", "target": "", "content": ""}]
|
||
out, reports = apply_patch_with_report("body", edits)
|
||
assert reports[0]["status"] == "skipped_missing_target"
|
||
|
||
def test_report_index_is_1_based(self) -> None:
|
||
edits = [
|
||
{"op": "append", "target": "", "content": "a"},
|
||
{"op": "append", "target": "", "content": "b"},
|
||
{"op": "append", "target": "", "content": "c"},
|
||
]
|
||
_, reports = apply_patch_with_report("body", edits)
|
||
assert [r["index"] for r in reports] == [1, 2, 3]
|
||
|
||
def test_report_truncation(self) -> None:
|
||
long_target = "x" * 300
|
||
long_content = "y" * 300
|
||
edits = [{"op": "replace", "target": long_target, "content": long_content}]
|
||
_, reports = apply_patch_with_report(long_target, edits)
|
||
assert len(reports[0]["target"]) == 200
|
||
assert len(reports[0]["content_preview"]) == 200
|
||
|
||
def test_ranges_recalculated_each_edit(self) -> None:
|
||
"""冻结区坐标在每条 edit 后重新计算。"""
|
||
protected = "FREEZE"
|
||
content = f"AAA\n{protected}\nBBB"
|
||
edits = [
|
||
{"op": "append", "target": "", "content": "prefix text"},
|
||
{"op": "replace", "target": protected, "content": "nope"},
|
||
]
|
||
out, reports = apply_patch_with_report(content, edits, protected_spans=[protected])
|
||
# append 在 FREEZE 之前插入，坐标右移后 replace 仍能检测冻结区
|
||
assert reports[1]["status"] == "skipped_protected"
|
||
|
||
def test_append_before_earliest_protected(self) -> None:
|
||
"""append 插到 start>0 的最早冻结区之前。"""
|
||
content = f"---\nfrontmatter\n---\n\nbody\n\n{APPENDIX_START}\nold\n{APPENDIX_END}"
|
||
edits = [{"op": "append", "target": "", "content": "INSERTED"}]
|
||
out, reports = apply_patch_with_report(
|
||
content, edits, protected_spans=[f"{APPENDIX_START}\nold\n{APPENDIX_END}"]
|
||
)
|
||
app_pos = out.find(APPENDIX_START)
|
||
ins_pos = out.find("INSERTED")
|
||
assert ins_pos < app_pos, "append 应在冻结区之前"
|
||
|
||
def test_payload_stripped_target_not_stripped(self) -> None:
|
||
"""target 不 strip，payload 做 strip。"""
|
||
content = " spaced target \nrest"
|
||
edits = [
|
||
{
|
||
"op": "replace",
|
||
"target": " spaced target ",
|
||
"content": " trimmed ",
|
||
}
|
||
]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
# payload stripped → "trimmed"
|
||
assert "trimmed" in out
|
||
assert " trimmed " not in out
|
||
assert reports[0]["status"] == "applied_replace"
|
||
|
||
def test_replace_count_one(self) -> None:
|
||
"""replace 只替换第一次出现。"""
|
||
content = "dup\ndup\ndup"
|
||
edits = [{"op": "replace", "target": "dup", "content": "REPLACED"}]
|
||
out, _ = apply_patch_with_report(content, edits)
|
||
assert out.count("REPLACED") == 1
|
||
assert out.count("dup") == 2
|
||
|
||
def test_multiple_edits_sequential(self) -> None:
|
||
"""多条 edit 顺序执行。"""
|
||
content = "line1\nline2\nline3"
|
||
edits = [
|
||
{"op": "replace", "target": "line1", "content": "LINE_ONE"},
|
||
{"op": "delete", "target": "line2", "content": ""},
|
||
{"op": "append", "target": "", "content": "TAIL"},
|
||
]
|
||
out, reports = apply_patch_with_report(content, edits)
|
||
assert "LINE_ONE" in out
|
||
assert "line2" not in out
|
||
assert "TAIL" in out
|
||
assert all(r["status"].startswith("applied") for r in reports)
|
||
|
||
|
||
class TestSpanOverlapProtection:
|
||
"""冻结区跨度保护 + marker 注入拦截（算法 #8 加固）。"""
|
||
|
||
def test_replace_spanning_into_protected_marker_preserved(self) -> None:
|
||
"""target 末端跨入含 marker 的 appendix 冻结区时被拦截，marker 不被破坏。"""
|
||
body = "正文最后一段。"
|
||
appendix = f"{APPENDIX_START}\n## 执行提醒\n- 规则A\n{APPENDIX_END}"
|
||
content = body + "\n\n" + appendix
|
||
# target 从正文末尾跨入 APPENDIX_START（含 marker 字面量）
|
||
target = "正文最后一段。\n\n" + APPENDIX_START
|
||
edits = [{"op": "delete", "target": target, "content": ""}]
|
||
new_content, report = apply_patch_with_report(
|
||
content, edits, protected_spans=[appendix]
|
||
)
|
||
# 无论经 marker 注入拦截还是跨度拦截，marker 都必须完整保留
|
||
assert APPENDIX_START in new_content and APPENDIX_END in new_content
|
||
assert report[0]["status"] in ("skipped_protected", "skipped_marker_injection")
|
||
|
||
def test_replace_spanning_into_frozen_section_is_skipped(self) -> None:
|
||
"""target 起点在正文、末端伸入无 marker 的冻结区段 → 跨度拦截跳过。"""
|
||
body = "可改正文段落。"
|
||
frozen = "## 输出格式\n必须输出 JSON。"
|
||
content = body + "\n" + frozen
|
||
# target 从可改正文跨入冻结区段（不含任何 marker 字面量）
|
||
target = "可改正文段落。\n## 输出格式"
|
||
edits = [{"op": "replace", "target": target, "content": "破坏内容"}]
|
||
new_content, report = apply_patch_with_report(
|
||
content, edits, protected_spans=[frozen]
|
||
)
|
||
assert "破坏内容" not in new_content
|
||
assert "## 输出格式" in new_content
|
||
assert report[0]["status"] == "skipped_protected"
|
||
|
||
def test_insert_after_spanning_into_frozen_section_is_skipped(self) -> None:
|
||
"""insert_after 的 target 末端伸入无 marker 冻结区段时跨度拦截跳过。"""
|
||
body = "可改正文。"
|
||
frozen = "## 输出格式\n固定内容。"
|
||
content = body + "\n" + frozen
|
||
target = "可改正文。\n## 输出格式"
|
||
edits = [{"op": "insert_after", "target": target, "content": "注入内容"}]
|
||
new_content, report = apply_patch_with_report(
|
||
content, edits, protected_spans=[frozen]
|
||
)
|
||
assert "注入内容" not in new_content
|
||
assert report[0]["status"] == "skipped_protected"
|
||
|
||
def test_edit_payload_with_marker_literal_rejected(self) -> None:
|
||
"""edit payload 含 marker 字面量 → 拒绝该 edit。"""
|
||
edits = [
|
||
{"op": "append", "target": "", "content": f"注入 {APPENDIX_START} 破坏"}
|
||
]
|
||
_, report = apply_patch_with_report("正文", edits, protected_spans=[])
|
||
assert any(
|
||
"marker" in str(s).lower() or "reject" in str(s).lower() for s in report[0].values()
|
||
)
|
||
|
||
def test_edit_target_with_marker_literal_rejected(self) -> None:
|
||
"""edit target 含 momentum marker 字面量 → 拒绝该 edit。"""
|
||
edits = [
|
||
{"op": "replace", "target": f"{MOMENTUM_START}x", "content": "y"}
|
||
]
|
||
_, report = apply_patch_with_report("正文", edits, protected_spans=[])
|
||
assert report[0]["status"] == "skipped_marker_injection"
|