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>
254 lines
9.0 KiB
Python
Executable File
254 lines
9.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Do Mode 功能测试"""
|
||
|
||
import sys
|
||
import os
|
||
import tempfile
|
||
import json
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent / "lib"))
|
||
|
||
def test_do_enter_worker():
|
||
"""测试 Worker 进入"""
|
||
from air_runtime.modes.do_mode import enter_worker
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
result = enter_worker(project_root, "T-001")
|
||
|
||
assert result["taskId"] == "T-001"
|
||
# 检查 worker_state 文件
|
||
from air_runtime.io import safe_json_load
|
||
worker_state_path = Path(result["workerStatePath"])
|
||
worker_state = safe_json_load(worker_state_path)
|
||
assert worker_state["status"] == "implementing"
|
||
|
||
print("✓ Worker 进入测试通过")
|
||
return True
|
||
|
||
|
||
def test_do_finish_worker():
|
||
"""测试 Worker 完成"""
|
||
from air_runtime.modes.do_mode import enter_worker, finish_worker
|
||
from air_runtime.contracts import WorkerResult
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
# 进入 worker
|
||
enter_worker(project_root, "T-001")
|
||
|
||
# 创建结果文件（Python 文件变更，不触发 C++ 检测）
|
||
result = WorkerResult(
|
||
task_id="T-001",
|
||
status="done",
|
||
summary="任务完成",
|
||
validations=[{"kind": "test", "status": "passed"}],
|
||
files_changed=["src/main.py"],
|
||
)
|
||
|
||
result_path = project_root / "AirPlan" / "state" / "airdo" / "tasks" / "T-001" / "result.json"
|
||
result_path.parent.mkdir(parents=True, exist_ok=True)
|
||
import json
|
||
result_path.write_text(json.dumps(result.to_dict()))
|
||
|
||
# 完成 worker
|
||
finish_result = finish_worker(project_root, "T-001")
|
||
|
||
assert finish_result["status"] == "done"
|
||
# 有证据 + Python 文件 → Dbg/XDB/NDB/SDB 都不触发，但 Rvr 必须触发
|
||
targets = [d["target"] for d in finish_result["routingDecisions"]]
|
||
assert "airrvr" in targets, f"所有 done 任务必须路由到 Rvr，实际: {targets}"
|
||
# 主路由决策是 airrvr（列表最后一个，因为是 append 的）
|
||
assert finish_result["routingDecision"]["target"] == "airrvr"
|
||
|
||
print(f"✓ Worker 完成测试通过，路由: {targets}")
|
||
return True
|
||
|
||
|
||
def test_do_ui_task_detection():
|
||
"""测试 UI 任务检测"""
|
||
from air_runtime.modes.do_mode import is_ui_task, route_ui_task
|
||
|
||
# 测试 is_ui_task
|
||
assert is_ui_task("实现登录界面UI") == True
|
||
assert is_ui_task("编写 React 组件") == True
|
||
assert is_ui_task("实现后端 API") == False
|
||
|
||
# 测试 route_ui_task（不实际安装 skill）
|
||
# 由于 skill 不存在，会返回 blocked
|
||
# 这里只测试函数能正常执行
|
||
result = route_ui_task("实现登录界面", "T-001")
|
||
assert "is_ui_task" in result
|
||
assert result["is_ui_task"] == True
|
||
|
||
print("✓ UI 任务检测测试通过")
|
||
return True
|
||
|
||
|
||
def test_do_force_airdbg():
|
||
"""测试强制 AirDbg 路由"""
|
||
from air_runtime.modes.do_mode import finish_worker
|
||
from air_runtime.contracts import WorkerResult
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
# 测试 done 但无证据 → 强制 AirDbg
|
||
result = WorkerResult(
|
||
task_id="T-002",
|
||
status="done",
|
||
summary="完成",
|
||
validations=[], # 无证据
|
||
files_changed=[], # 无文件变更
|
||
)
|
||
|
||
result_path = project_root / "AirPlan" / "state" / "airdo" / "tasks" / "T-002" / "result.json"
|
||
result_path.parent.mkdir(parents=True, exist_ok=True)
|
||
import json
|
||
result_path.write_text(json.dumps(result.to_dict()))
|
||
|
||
finish_result = finish_worker(project_root, "T-002")
|
||
|
||
# 应该强制路由到 airdbg（无证据）+ airrvr（所有 done）
|
||
targets = [d["target"] for d in finish_result["routingDecisions"]]
|
||
assert "airdbg" in targets, f"无证据 done 必须路由到 Dbg，实际: {targets}"
|
||
assert "airrvr" in targets, f"所有 done 必须路由到 Rvr，实际: {targets}"
|
||
assert finish_result["routingDecision"]["forced"] == True
|
||
|
||
print(f"✓ 强制 AirDbg 路由测试通过，路由: {targets}")
|
||
return True
|
||
|
||
|
||
def test_do_sanitize_task_id():
|
||
"""测试 task_id 注入防护"""
|
||
from air_runtime.utils import sanitize_task_id
|
||
|
||
# 正常 ID
|
||
assert sanitize_task_id("T-001") == "T-001"
|
||
|
||
# 危险字符应该抛出异常（而非清理后放行）
|
||
try:
|
||
sanitize_task_id("../../../etc/passwd")
|
||
assert False, "应该抛出异常"
|
||
except ValueError:
|
||
pass # 正确行为：拒绝危险输入
|
||
|
||
print("✓ task_id 注入防护测试通过")
|
||
return True
|
||
|
||
|
||
def test_do_blocked_routes_to_dbg():
|
||
"""测试 blocked 状态强制路由到 AirDbg"""
|
||
from air_runtime.modes.do_mode import finish_worker
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
result_data = {
|
||
"taskId": "T-B01", "status": "blocked",
|
||
"summary": "编译失败", "filesChanged": ["src/main.cpp"],
|
||
}
|
||
result_path = project_root / "AirPlan" / "state" / "airdo" / "tasks" / "T-B01" / "result.json"
|
||
result_path.parent.mkdir(parents=True, exist_ok=True)
|
||
result_path.write_text(json.dumps(result_data))
|
||
|
||
finish_result = finish_worker(project_root, "T-B01")
|
||
targets = [d["target"] for d in finish_result["routingDecisions"]]
|
||
assert "airdbg" in targets, f"blocked 必须路由到 Dbg，实际: {targets}"
|
||
print(f"✓ blocked→Dbg 路由测试通过，路由: {targets}")
|
||
return True
|
||
|
||
|
||
def test_do_cpp_routes_to_sdb():
|
||
"""测试 C/C++ 任务强制路由到 AirSDB"""
|
||
from air_runtime.modes.do_mode import finish_worker
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
result_data = {
|
||
"taskId": "T-C01", "status": "done",
|
||
"summary": "完成", "validations": [{"kind": "build", "status": "passed"}],
|
||
"filesChanged": ["src/decoder.cpp", "include/decoder.h"],
|
||
}
|
||
result_path = project_root / "AirPlan" / "state" / "airdo" / "tasks" / "T-C01" / "result.json"
|
||
result_path.parent.mkdir(parents=True, exist_ok=True)
|
||
result_path.write_text(json.dumps(result_data))
|
||
|
||
finish_result = finish_worker(project_root, "T-C01")
|
||
targets = [d["target"] for d in finish_result["routingDecisions"]]
|
||
assert "airsdb" in targets, f"C/C++ 任务必须路由到 SDB，实际: {targets}"
|
||
assert "airrvr" in targets, f"done 任务必须路由到 Rvr，实际: {targets}"
|
||
print(f"✓ C++→SDB 路由测试通过，路由: {targets}")
|
||
return True
|
||
|
||
|
||
def test_do_done_with_evidence_routes_to_merge_or_rvr():
|
||
"""测试 done + 有证据 → 只路由到 Rvr（不触发 Dbg）"""
|
||
from air_runtime.modes.do_mode import finish_worker
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
result_data = {
|
||
"taskId": "T-D01", "status": "done",
|
||
"summary": "完成",
|
||
"validations": [{"kind": "test", "status": "passed"}],
|
||
"filesChanged": ["src/utils.py"],
|
||
}
|
||
result_path = project_root / "AirPlan" / "state" / "airdo" / "tasks" / "T-D01" / "result.json"
|
||
result_path.parent.mkdir(parents=True, exist_ok=True)
|
||
result_path.write_text(json.dumps(result_data))
|
||
|
||
finish_result = finish_worker(project_root, "T-D01")
|
||
targets = [d["target"] for d in finish_result["routingDecisions"]]
|
||
# 有验证证据 → 不触发 Dbg
|
||
assert "airdbg" not in targets, f"有证据不应触发 Dbg，实际: {targets}"
|
||
# done → 必须触发 Rvr
|
||
assert "airrvr" in targets, f"done 必须路由到 Rvr，实际: {targets}"
|
||
print(f"✓ 有证据→无Dbg+有Rvr 测试通过，路由: {targets}")
|
||
return True
|
||
|
||
|
||
def main():
|
||
print("=" * 50)
|
||
print("Do Mode 功能测试")
|
||
print("=" * 50)
|
||
|
||
tests = [
|
||
("Worker 进入", test_do_enter_worker),
|
||
("Worker 完成", test_do_finish_worker),
|
||
("UI 任务检测", test_do_ui_task_detection),
|
||
("强制 AirDbg 路由", test_do_force_airdbg),
|
||
("task_id 注入防护", test_do_sanitize_task_id),
|
||
("blocked→Dbg 路由", test_do_blocked_routes_to_dbg),
|
||
("C++→SDB 路由", test_do_cpp_routes_to_sdb),
|
||
("有证据→无Dbg+有Rvr", test_do_done_with_evidence_routes_to_merge_or_rvr),
|
||
]
|
||
|
||
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) |