test: 全模式功能测试覆盖 (12模式 33项测试)
新增 arc/eng/do/dbg/ctx/xdb/sdb/ndb/dep/tst/sec/rvr 全模式功能测试; fix do_mode UI 检测中文关键词补全;scripts/install.sh 同步更新。 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
172
test_do.py
Normal file
172
test_do.py
Normal file
@@ -0,0 +1,172 @@
|
||||
#!/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")
|
||||
|
||||
# 创建结果文件
|
||||
result = WorkerResult(
|
||||
task_id="T-001",
|
||||
status="done",
|
||||
summary="任务完成",
|
||||
validations="编译通过",
|
||||
files_changed=["src/main.cpp"],
|
||||
)
|
||||
|
||||
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"
|
||||
assert finish_result["routingDecision"]["target"] in ["merge", "airdbg"]
|
||||
|
||||
print(f"✓ Worker 完成测试通过,路由: {finish_result['routingDecision']['target']}")
|
||||
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
|
||||
assert finish_result["routingDecision"]["target"] == "airdbg"
|
||||
assert finish_result["routingDecision"]["forced"] == True
|
||||
|
||||
print("✓ 强制 AirDbg 路由测试通过")
|
||||
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 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),
|
||||
]
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user