a60d1a0c04
- lock.py: 跨平台进程锁（Unix fcntl / Windows msvcrt / O_CREAT|O_EXCL降级）
- eng_mode.py/eng_orchestrator.py: hasattr(os, "getloadavg") Windows防护
- arc_mode.py: 路径分隔符 replace("\\", "/") Windows兼容
- deploy_runtime.py: 修复语法错误（清理 import tempfile 残留）
- P1-24(3.2.18): AMBIGUOUS_VERBS歧义词检测 + SAFE_VERBS安全动词 + validate_task_description()
- TaskNode.keep_constraints 保留约束字段 + JSON序列化
- _build_graph_from_todo 返回歧义警告 + Arc自检集成
- 3.2.9b: FORBIDDEN_DEGRADATION_PATTERNS + check_forbidden_degradation()
- AirRvr三层审查放行标准: ReviewVerdict + evaluate_review_pass() + is_forbidden_pass_reason()
- commands/arc.md: 弱模型安全重写（操作类型拆分+保留约束+自检）
- commands/do.md/eng.md/rvr.md: 禁止降级方案 + 三层审查标准
- 测试: 7个新测试 + 74全量通过
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
295 lines
11 KiB
Python
Executable File
295 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Arc Mode 功能测试"""
|
||
|
||
import sys
|
||
import os
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent / "lib"))
|
||
|
||
def test_arc_phase_gate():
|
||
"""测试三阶段门控"""
|
||
from air_runtime.modes.arc_mode import ArcPhaseGate
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
state_path = Path(tmpdir) / "state.json"
|
||
gate = ArcPhaseGate(state_path)
|
||
|
||
# 初始阶段应该是 discussing
|
||
assert gate.current_phase == "discussing", f"初始阶段应为 discussing，实际: {gate.current_phase}"
|
||
|
||
# 可以前进到 proposing
|
||
gate.advance_to("proposing")
|
||
assert gate.current_phase == "proposing"
|
||
|
||
# 不能跳到 confirmed（只能逐步前进）
|
||
gate.advance_to("confirmed")
|
||
assert gate.current_phase == "confirmed"
|
||
|
||
# can_write_plan 只在 confirmed 时返回 True
|
||
assert gate.can_write_plan() == True
|
||
|
||
# 测试确认架构
|
||
result = gate.confirm_architecture("好的，确认这个架构")
|
||
assert result == True
|
||
assert gate.current_phase == "confirmed"
|
||
|
||
print("✓ ArcPhaseGate 三阶段门控测试通过")
|
||
return True
|
||
|
||
|
||
def test_arc_build_graph():
|
||
"""测试 DAG 构建"""
|
||
from air_runtime.modes.arc_mode import _build_graph_from_todo
|
||
from air_runtime.task_graph import TaskGraph
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
todo_path = Path(tmpdir) / "todo.md"
|
||
todo_path.write_text("""
|
||
| Task | Status | Files/Dirs | Done When |
|
||
|------|--------|------------|-----------|
|
||
| [T-001] 任务1 | TODO | src/a/ | 完成 |
|
||
| [T-002] 任务2 | TODO | src/b/ | 完成 |
|
||
| [T-003] 任务3 | DONE | src/c/ | 完成 |
|
||
""")
|
||
|
||
graph, violations, ambiguity_warnings = _build_graph_from_todo(todo_path)
|
||
|
||
# 检查节点数量
|
||
assert len(graph.nodes) >= 3, f"应有至少3个节点，实际: {len(graph.nodes)}"
|
||
|
||
# 检查是否有边界测试任务注入
|
||
test_nodes = [n for n in graph.nodes.values() if n.test_required]
|
||
assert len(test_nodes) > 0, "应该有边界测试任务"
|
||
|
||
# 检查 Done When 违规检测
|
||
assert len(violations) > 0, "应该有 Done When 不含'测试通过'的违规"
|
||
|
||
print(f"✓ DAG 构建测试通过，节点数: {len(graph.nodes)}, 测试任务: {len(test_nodes)}, 违规: {len(violations)}, 歧义警告: {len(ambiguity_warnings)}")
|
||
return True
|
||
|
||
|
||
def test_arc_parallel_review():
|
||
"""测试并行审查模式"""
|
||
from air_runtime.modes.arc_mode import parallel_review_mode
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
project_root = Path(tmpdir)
|
||
|
||
# 创建 AirPlan 目录结构
|
||
arc_root = project_root / "AirPlan" / "state" / "airarc"
|
||
arc_root.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 写入 state.json（需先有 confirmed 阶段）
|
||
import json
|
||
(arc_root / "state.json").write_text(json.dumps({"arcPhase": "confirmed"}))
|
||
|
||
# 创建 todo.md
|
||
todo_path = project_root / "AirPlan" / "todo.md"
|
||
todo_path.write_text("""
|
||
| Task | Status | Files/Dirs | Done When |
|
||
|------|--------|------------|-----------|
|
||
| [T-001] 任务1 | TODO | src/main.cpp | 编译通过 |
|
||
""")
|
||
|
||
result = parallel_review_mode(project_root, todo_path)
|
||
|
||
# 检查返回值
|
||
assert "execution_plan_json_path" in result
|
||
assert result.get("blocked") != True, "不应被阻止"
|
||
|
||
# 检查 execution-plan.json 是否生成
|
||
plan_path = result["execution_plan_json_path"]
|
||
assert Path(plan_path).exists(), f"执行计划文件应存在: {plan_path}"
|
||
|
||
print(f"✓ parallel_review_mode 测试通过")
|
||
return True
|
||
|
||
|
||
def test_validate_task_description():
|
||
"""P1-24: 测试任务描述歧义词检测"""
|
||
from air_runtime.review import validate_task_description
|
||
|
||
# 包含歧义词
|
||
warnings = validate_task_description("T-001", "清理旧产品实现并重建新CMake骨架")
|
||
assert len(warnings) > 0, "应该检测到歧义词'清理'"
|
||
assert "清理" in warnings[0], f"警告应包含'清理': {warnings[0]}"
|
||
|
||
# 不包含歧义词
|
||
clean_warnings = validate_task_description("T-002", "重构CMakeLists.txt去掉sipclient依赖")
|
||
assert len(clean_warnings) == 0, f"不应该有歧义词警告: {clean_warnings}"
|
||
|
||
# 多个歧义词
|
||
multi_warnings = validate_task_description("T-003", "清理旧代码并优化模块结构，更新API接口")
|
||
assert len(multi_warnings) >= 3, f"应该检测到至少3个歧义词: {len(multi_warnings)}"
|
||
|
||
print(f"✓ validate_task_description 测试通过，歧义词检测: {len(warnings)}/{len(clean_warnings)}/{len(multi_warnings)}")
|
||
return True
|
||
|
||
|
||
def test_ambiguity_warnings_in_graph():
|
||
"""P1-24: 测试 _build_graph_from_todo 产出歧义警告"""
|
||
from air_runtime.modes.arc_mode import _build_graph_from_todo
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
todo_path = Path(tmpdir) / "todo.md"
|
||
todo_path.write_text("""
|
||
| Task | Status | Files/Dirs | Done When |
|
||
|------|--------|------------|-----------|
|
||
| [T-001] 清理旧模块 | TODO | src/module.cpp | 编译通过 |
|
||
| [T-002] 优化数据库查询性能 | TODO | db/query.cpp | 测试通过 |
|
||
| [T-003] 新增日志模块 | TODO | log/spdlog.cpp | 测试通过 |
|
||
""")
|
||
graph, violations, ambiguity_warnings = _build_graph_from_todo(todo_path)
|
||
|
||
# T-001 "清理" 和 T-002 "优化" 应该触发歧义警告
|
||
assert len(ambiguity_warnings) >= 2, f"应有至少2个歧义警告，实际: {len(ambiguity_warnings)}"
|
||
# T-003 "新增" 应该没有歧义
|
||
ambiguous_task_ids = set()
|
||
for w in ambiguity_warnings:
|
||
tid = w.split("]")[0].replace("[", "") if "]" in w else ""
|
||
ambiguous_task_ids.add(tid)
|
||
assert "T-003" not in ambiguous_task_ids, f"'新增'不应触发歧义警告: {ambiguity_warnings}"
|
||
|
||
print(f"✓ 歧义警告产出测试通过，警告: {len(ambiguity_warnings)}，违规: {len(violations)}")
|
||
return True
|
||
|
||
|
||
def test_forbidden_degradation():
|
||
"""3.2.9b: 测试降级语言检测"""
|
||
from air_runtime.review import check_forbidden_degradation
|
||
|
||
# 包含降级语言
|
||
found = check_forbidden_degradation("我们先用兜底方案实现，以后再优化")
|
||
assert len(found) > 0, "应该检测到降级语言"
|
||
assert "兜底方案" in found or "以后再优化" in found, f"应检测到具体模式: {found}"
|
||
|
||
# 不包含降级语言
|
||
clean = check_forbidden_degradation("按照AirArc设计方案实现，使用RAII管理资源")
|
||
assert len(clean) == 0, f"不应有降级语言: {clean}"
|
||
|
||
# 多种降级语言
|
||
multi = check_forbidden_degradation("先用临时方案实现，以后再删掉这个兜底方案")
|
||
assert len(multi) >= 3, f"应检测到多个降级模式: {len(multi)}"
|
||
|
||
print(f"✓ 降级语言检测测试通过: {len(found)}/{len(clean)}/{len(multi)}")
|
||
return True
|
||
|
||
|
||
def test_evaluate_review_pass():
|
||
"""AirRvr: 测试三层审查放行标准"""
|
||
from air_runtime.review import evaluate_review_pass, is_forbidden_pass_reason
|
||
|
||
# 全部通过
|
||
result = evaluate_review_pass(
|
||
code_to_design_table=[
|
||
{"designPoint": "使用RAII管理资源", "status": "aligned"},
|
||
{"designPoint": "接口使用智能指针", "status": "aligned"},
|
||
],
|
||
sdb_report={"findings": []},
|
||
test_results={"suites": [{"name": "unit", "status": "pass"}]},
|
||
)
|
||
assert result.verdict == "pass", f"全部通过应为 pass，实际: {result.verdict}"
|
||
assert result.layer1_pass and result.layer2_pass and result.layer3_pass
|
||
|
||
# 第一层有 divergent → fail
|
||
result2 = evaluate_review_pass(
|
||
code_to_design_table=[
|
||
{"designPoint": "使用RAII管理资源", "status": "divergent"},
|
||
],
|
||
sdb_report={"findings": []},
|
||
test_results={"suites": [{"name": "unit", "status": "pass"}]},
|
||
)
|
||
assert result2.verdict == "fail", f"第一层有divergent应为fail，实际: {result2.verdict}"
|
||
assert not result2.layer1_pass
|
||
|
||
# 第一层通过，第二层有 critical finding → conditional-pass
|
||
result3 = evaluate_review_pass(
|
||
code_to_design_table=[
|
||
{"designPoint": "使用RAII管理资源", "status": "aligned"},
|
||
],
|
||
sdb_report={"findings": [{"severity": "critical", "description": "use-after-free"}]},
|
||
test_results={"suites": [{"name": "unit", "status": "pass"}]},
|
||
)
|
||
assert result3.verdict == "conditional-pass", f"应为conditional-pass，实际: {result3.verdict}"
|
||
|
||
# 第一层有 missing → fail
|
||
result4 = evaluate_review_pass(
|
||
code_to_design_table=[
|
||
{"designPoint": "使用RAII管理资源", "status": "aligned"},
|
||
{"designPoint": "AES加密", "status": "missing"},
|
||
],
|
||
sdb_report={"findings": []},
|
||
test_results={"suites": []},
|
||
)
|
||
assert result4.verdict == "fail", f"第一层有missing应为fail，实际: {result4.verdict}"
|
||
|
||
# 禁止的表面原因
|
||
assert is_forbidden_pass_reason("测试全绿") == True
|
||
assert is_forbidden_pass_reason("编译通过无报错") == True
|
||
assert is_forbidden_pass_reason("代码实现正确，三层审查均通过") == False
|
||
|
||
print(f"✓ 三层审查放行标准测试通过: pass/fail/cond-pass/forbidden")
|
||
return True
|
||
|
||
|
||
def test_keep_constraints_in_tasknode():
|
||
"""P1-24: 测试 TaskNode.keep_constraints 字段"""
|
||
from air_runtime.task_graph import TaskNode
|
||
|
||
node = TaskNode(
|
||
id="T-001",
|
||
task="重构 CMakeLists.txt 去掉 sipclient 依赖",
|
||
keep_constraints=["src/ 目录下所有现有源文件不得删除或修改"],
|
||
)
|
||
assert node.keep_constraints == ["src/ 目录下所有现有源文件不得删除或修改"]
|
||
assert len(node.keep_constraints) == 1
|
||
|
||
# 空约束
|
||
node2 = TaskNode(id="T-002", task="新增日志模块")
|
||
assert node2.keep_constraints == []
|
||
|
||
print("✓ TaskNode.keep_constraints 字段测试通过")
|
||
return True
|
||
|
||
|
||
def main():
|
||
print("=" * 50)
|
||
print("Arc Mode 功能测试")
|
||
print("=" * 50)
|
||
|
||
tests = [
|
||
("三阶段门控", test_arc_phase_gate),
|
||
("DAG 构建", test_arc_build_graph),
|
||
("并行审查", test_arc_parallel_review),
|
||
("P1-24 歧义词检测", test_validate_task_description),
|
||
("P1-24 歧义警告产出", test_ambiguity_warnings_in_graph),
|
||
("3.2.9b 降级语言检测", test_forbidden_degradation),
|
||
("三层审查放行标准", test_evaluate_review_pass),
|
||
("P1-24 keep_constraints", test_keep_constraints_in_tasknode),
|
||
]
|
||
|
||
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) |