ae44be31d5
Includes AirPlan design documents, AircOding-alpha1-plan, AirPlanV2, AirPlan-ParaV2, AirPlan-Para V1 reference docs, and all working code changes across packages. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
359 lines
13 KiB
Python
Executable File
359 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""P1-21 补充功能测试：ADRWatcher + PartialReplanner"""
|
||
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent / "lib"))
|
||
|
||
|
||
def test_adr_watcher_snapshot():
|
||
"""测试 ADRWatcher 初始快照"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
adr_dir = Path(tmpdir)
|
||
# 创建 ADR 文件
|
||
(adr_dir / "ADR-0005-ffmpeg-decode.md").write_text(
|
||
"# ADR-0005: ffmpeg 解码\n\nStatus: accepted\n\n使用 ffmpeg 实现视频解码。"
|
||
)
|
||
(adr_dir / "ADR-0012-auth.md").write_text(
|
||
"# ADR-0012: 认证方案\n\nStatus: accepted\n\n使用 JWT 认证。"
|
||
)
|
||
|
||
watcher = ADRWatcher(adr_dir)
|
||
watcher.snapshot()
|
||
|
||
assert len(watcher._known_hashes) == 2
|
||
assert "ADR-0005" in watcher._known_hashes
|
||
assert "ADR-0012" in watcher._known_hashes
|
||
|
||
# 初始快照后无变更
|
||
changes = watcher.detect_changes()
|
||
assert len(changes) == 0
|
||
|
||
print("✓ ADRWatcher 初始快照测试通过")
|
||
return True
|
||
|
||
|
||
def test_adr_watcher_detect_modified():
|
||
"""测试 ADRWatcher 检测 ADR 修改"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
adr_dir = Path(tmpdir)
|
||
adr_file = adr_dir / "ADR-0005-ffmpeg-decode.md"
|
||
adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n用 ffmpeg。")
|
||
|
||
watcher = ADRWatcher(adr_dir)
|
||
watcher.snapshot()
|
||
|
||
# 修改 ADR 内容
|
||
adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n改用 gstreamer。")
|
||
changes = watcher.detect_changes()
|
||
|
||
assert len(changes) == 1
|
||
assert changes[0].adr_id == "ADR-0005"
|
||
assert changes[0].kind == "modified"
|
||
assert changes[0].old_hash != changes[0].new_hash
|
||
|
||
print("✓ ADRWatcher 检测修改测试通过")
|
||
return True
|
||
|
||
|
||
def test_adr_watcher_detect_superseded():
|
||
"""测试 ADRWatcher 检测 ADR 被替代"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
adr_dir = Path(tmpdir)
|
||
adr_file = adr_dir / "ADR-0005-ffmpeg-decode.md"
|
||
adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n用 ffmpeg。")
|
||
|
||
watcher = ADRWatcher(adr_dir)
|
||
watcher.snapshot()
|
||
|
||
# 标记为 superseded
|
||
adr_file.write_text("# ADR-0005\n\nStatus: superseded\n\n被 ADR-0020 替代。")
|
||
changes = watcher.detect_changes()
|
||
|
||
assert len(changes) == 1
|
||
assert changes[0].kind == "superseded"
|
||
|
||
print("✓ ADRWatcher 检测 superseded 测试通过")
|
||
return True
|
||
|
||
|
||
def test_adr_watcher_detect_new():
|
||
"""测试 ADRWatcher 检测新增 ADR"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
adr_dir = Path(tmpdir)
|
||
(adr_dir / "ADR-0005-old.md").write_text("# ADR-0005\n\nStatus: accepted\n")
|
||
|
||
watcher = ADRWatcher(adr_dir)
|
||
watcher.snapshot()
|
||
|
||
# 新增 ADR
|
||
(adr_dir / "ADR-0020-gstreamer.md").write_text("# ADR-0020\n\nStatus: accepted\n")
|
||
changes = watcher.detect_changes()
|
||
|
||
assert len(changes) == 1
|
||
assert changes[0].adr_id == "ADR-0020"
|
||
assert changes[0].kind == "new"
|
||
|
||
print("✓ ADRWatcher 检测新增 ADR 测试通过")
|
||
return True
|
||
|
||
|
||
def test_adr_watcher_extract_id():
|
||
"""测试 ADR ID 提取"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
|
||
assert ADRWatcher._extract_adr_id(Path("ADR-0005-ffmpeg-decode.md")) == "ADR-0005"
|
||
assert ADRWatcher._extract_adr_id(Path("ADR-0012.md")) == "ADR-0012"
|
||
assert ADRWatcher._extract_adr_id(Path("ADR-9999-super-long-name.md")) == "ADR-9999"
|
||
|
||
print("✓ ADR ID 提取测试通过")
|
||
return True
|
||
|
||
|
||
def test_adr_watcher_parse_status():
|
||
"""测试 ADR Status 字段解析"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
f1 = Path(tmpdir) / "ADR-0001.md"
|
||
f1.write_text("# ADR-0001\n\nStatus: accepted\n")
|
||
assert ADRWatcher._parse_status(f1) == "accepted"
|
||
|
||
f2 = Path(tmpdir) / "ADR-0002.md"
|
||
f2.write_text("# ADR-0002\n\nStatus: superseded\n")
|
||
assert ADRWatcher._parse_status(f2) == "superseded"
|
||
|
||
f3 = Path(tmpdir) / "ADR-0003.md"
|
||
f3.write_text("# ADR-0003\n\n无 Status 字段\n")
|
||
assert ADRWatcher._parse_status(f3) == "unknown"
|
||
|
||
print("✓ ADR Status 解析测试通过")
|
||
return True
|
||
|
||
|
||
def test_partial_replanner():
|
||
"""测试局部重规划器"""
|
||
from air_runtime.task_graph import TaskGraph, TaskNode, Edge
|
||
from air_runtime.partial_replanner import PartialReplanner
|
||
|
||
graph = TaskGraph()
|
||
# 失效的任务
|
||
graph.add_node(TaskNode(id="T-001", status="INVALIDATED", task="ffmpeg 解码器",
|
||
files_dirs="src/decoder.cpp", write_set=["src/decoder.cpp"],
|
||
adr_refs=["ADR-0005"]))
|
||
graph.add_node(TaskNode(id="T-002", status="INVALIDATED", task="ffmpeg 编码器",
|
||
files_dirs="src/encoder.cpp", write_set=["src/encoder.cpp"],
|
||
adr_refs=["ADR-0005"]))
|
||
# 稳定的已完成任务
|
||
graph.add_node(TaskNode(id="T-003", status="DONE", task="UI 界面",
|
||
write_set=["src/ui/main.cpp"], adr_refs=["ADR-0008"]))
|
||
graph.add_node(TaskNode(id="T-004", status="DONE", task="日志模块",
|
||
write_set=["src/logger.cpp"], adr_refs=[]))
|
||
|
||
replanner = PartialReplanner()
|
||
delta = replanner.replan(graph, ["T-001", "T-002"])
|
||
|
||
# 检查 replan_request 中的受影响上下文
|
||
request = delta.replan_request
|
||
assert request["type"] == "partial-replan"
|
||
assert set(request["invalidatedTaskIds"]) == {"T-001", "T-002"}
|
||
assert len(request["affectedContext"]) == 2
|
||
assert request["affectedContext"][0]["task_id"] == "T-001"
|
||
|
||
# 检查稳定接口
|
||
stable = request["stableInterfaces"]
|
||
assert len(stable) == 2 # T-003 和 T-004
|
||
stable_ids = {s["task_id"] for s in stable}
|
||
assert "T-003" in stable_ids
|
||
assert "T-004" in stable_ids
|
||
|
||
print("✓ 局部重规划器测试通过")
|
||
return True
|
||
|
||
|
||
def test_partial_replanner_stable_interfaces():
|
||
"""测试稳定接口提取"""
|
||
from air_runtime.task_graph import TaskGraph, TaskNode
|
||
from air_runtime.partial_replanner import PartialReplanner
|
||
|
||
graph = TaskGraph()
|
||
graph.add_node(TaskNode(id="T-001", status="INVALIDATED", task="失效任务", adr_refs=["ADR-0005"]))
|
||
graph.add_node(TaskNode(id="T-002", status="DONE", task="完成有写集",
|
||
write_set=["src/a.cpp", "src/b.cpp"], adr_refs=["ADR-0008"]))
|
||
graph.add_node(TaskNode(id="T-003", status="DONE", task="完成无写集"))
|
||
graph.add_node(TaskNode(id="T-004", status="TODO", task="未开始"))
|
||
|
||
replanner = PartialReplanner()
|
||
interfaces = replanner._extract_stable_interfaces(graph, {"T-001"})
|
||
|
||
# T-002 有写集，应该包含
|
||
# T-003 无写集且无 adr_refs，不包含
|
||
# T-004 不是 DONE，不包含
|
||
assert len(interfaces) == 1
|
||
assert interfaces[0].task_id == "T-002"
|
||
|
||
print("✓ 稳定接口提取测试通过")
|
||
return True
|
||
|
||
|
||
def test_eng_monitor_detects_adr_change():
|
||
"""测试 Eng 监控循环检测 ADR 变更"""
|
||
from air_runtime.modes.eng_mode import enter_engine, monitor_engine, _detect_adr_changes
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
# 初始化引擎
|
||
enter_engine(project_root)
|
||
|
||
# 创建 ADR 目录（无文件）
|
||
adr_dir = project_root / "AirPlan" / "docs" / "architecture" / "adr"
|
||
adr_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 创建初始 ADR
|
||
(adr_dir / "ADR-0005-ffmpeg.md").write_text("# ADR-0005\n\nStatus: accepted\n\n用 ffmpeg。")
|
||
|
||
# 第一次监控：初始化 hash 快照
|
||
from air_runtime.io import safe_json_load
|
||
from air_runtime.modes.eng_mode import _paths
|
||
paths = _paths(project_root)
|
||
state = safe_json_load(paths["state"]) or {}
|
||
changes = _detect_adr_changes(project_root, state)
|
||
assert len(changes) == 0 # 首次只初始化，无变更
|
||
|
||
# 保存 hash 到 state
|
||
from air_runtime.io import atomic_json_write
|
||
atomic_json_write(paths["state"], state)
|
||
|
||
# 修改 ADR 为 superseded
|
||
(adr_dir / "ADR-0005-ffmpeg.md").write_text("# ADR-0005\n\nStatus: superseded\n\n被替代。")
|
||
|
||
# 第二次检测
|
||
state = safe_json_load(paths["state"]) or {}
|
||
changes = _detect_adr_changes(project_root, state)
|
||
assert len(changes) == 1
|
||
assert changes[0].adr_id == "ADR-0005"
|
||
assert changes[0].kind == "superseded"
|
||
|
||
print("✓ Eng 监控检测 ADR 变更测试通过")
|
||
return True
|
||
|
||
|
||
def test_full_adr_change_to_invalidation_flow():
|
||
"""完整场景：ADR 变更 → 自动检测 → 级联失效 → 局部重规划"""
|
||
from air_runtime.adr_watcher import ADRWatcher
|
||
from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta
|
||
from air_runtime.partial_replanner import PartialReplanner
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
adr_dir = project_root / "docs" / "architecture" / "adr"
|
||
adr_dir.mkdir(parents=True)
|
||
|
||
# 初始 ADR
|
||
adr_file = adr_dir / "ADR-0005-ffmpeg-decode.md"
|
||
adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n使用 ffmpeg 实现视频解码。")
|
||
|
||
# 初始化 watcher
|
||
watcher = ADRWatcher(adr_dir)
|
||
watcher.snapshot()
|
||
|
||
# 初始化 task graph
|
||
graph = TaskGraph()
|
||
graph.add_node(TaskNode(id="T-001", status="DONE", task="ffmpeg 解码器",
|
||
files_dirs="src/decoder.cpp", write_set=["src/decoder.cpp"],
|
||
adr_refs=["ADR-0005"]))
|
||
graph.add_node(TaskNode(id="T-002", status="TODO", task="解码器集成测试"))
|
||
graph.add_node(TaskNode(id="T-003", status="DONE", task="UI 界面",
|
||
write_set=["src/ui.cpp"], adr_refs=["ADR-0008"]))
|
||
graph.add_edge(Edge(source="T-001", target="T-002"))
|
||
|
||
# 用户修改 ADR（ffmpeg → gstreamer）
|
||
adr_file.write_text("# ADR-0005\n\nStatus: superseded\n\n改用 gstreamer。")
|
||
|
||
# Watcher 检测变更
|
||
changes = watcher.detect_changes()
|
||
assert len(changes) == 1
|
||
assert changes[0].kind == "superseded"
|
||
assert changes[0].adr_id == "ADR-0005"
|
||
|
||
# 触发级联失效
|
||
delta = PlanDelta()
|
||
report = graph.invalidate_by_adr("ADR-0005", delta)
|
||
assert report.invalidated_completed == 1
|
||
assert report.cascaded_downstream == 1
|
||
assert graph.dispatch_frozen == True
|
||
|
||
# 局部重规划
|
||
replanner = PartialReplanner()
|
||
partial_delta = replanner.replan(graph, report.invalidated_task_ids)
|
||
request = partial_delta.replan_request
|
||
assert request["type"] == "partial-replan"
|
||
assert "T-001" in request["invalidatedTaskIds"]
|
||
assert "T-002" in request["invalidatedTaskIds"]
|
||
# T-003 是稳定的
|
||
assert len(request["stableInterfaces"]) == 1
|
||
assert request["stableInterfaces"][0]["task_id"] == "T-003"
|
||
|
||
# 解冻
|
||
graph.add_node(TaskNode(id="T-101", status="TODO", task="gstreamer 解码器",
|
||
files_dirs="src/decoder.cpp", adr_refs=["ADR-0005"]))
|
||
graph.unfreeze_dispatch()
|
||
assert graph.dispatch_frozen == False
|
||
|
||
print("✓ 完整 ADR 变更流程测试通过（检测→失效→重规划→解冻）")
|
||
return True
|
||
|
||
|
||
def main():
|
||
print("=" * 50)
|
||
print("P1-21 补充功能测试：ADRWatcher + PartialReplanner")
|
||
print("=" * 50)
|
||
|
||
tests = [
|
||
("ADRWatcher 快照", test_adr_watcher_snapshot),
|
||
("ADRWatcher 检测修改", test_adr_watcher_detect_modified),
|
||
("ADRWatcher 检测 superseded", test_adr_watcher_detect_superseded),
|
||
("ADRWatcher 检测新增", test_adr_watcher_detect_new),
|
||
("ADR ID 提取", test_adr_watcher_extract_id),
|
||
("ADR Status 解析", test_adr_watcher_parse_status),
|
||
("局部重规划器", test_partial_replanner),
|
||
("稳定接口提取", test_partial_replanner_stable_interfaces),
|
||
("Eng 监控检测 ADR", test_eng_monitor_detects_adr_change),
|
||
("完整 ADR 流程", test_full_adr_change_to_invalidation_flow),
|
||
]
|
||
|
||
passed = 0
|
||
failed = 0
|
||
|
||
for name, test_fn in tests:
|
||
try:
|
||
test_fn()
|
||
passed += 1
|
||
except Exception as e:
|
||
print(f"✗ {name} 失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
failed += 1
|
||
|
||
print("=" * 50)
|
||
print(f"测试结果: {passed} 通过, {failed} 失败")
|
||
print("=" * 50)
|
||
|
||
return failed == 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
success = main()
|
||
sys.exit(0 if success else 1)
|