直接运行简化测试,无需任何外部依赖:
python test_simple.py这将测试所有核心功能:
- ✅ Coder 节点(代码生成)
- ✅ Debugger 节点(代码调试)
- ✅ Checker 节点(代码验证)
- ✅ Tool 节点(代码执行)
🧪 ==========================================================
🧪 LangGraph 代码助手 - 简化测试套件
🧪 ==========================================================
============================================================
测试 1: 单个 Coder 节点
============================================================
📊 测试结果:
错误状态: no
迭代次数: 1
代码前缀: 生成的代码解决方案
代码内容: def example_function():
return 'Hello from Coder!'
✅ 测试通过
============================================================
测试 2: 所有节点顺序执行
============================================================
📝 初始需求: 实现一个 RAG 链
步骤 1/4: Coder 生成代码
步骤 2/4: Debugger 调试代码
步骤 3/4: Checker 验证代码
步骤 4/4: Tool 执行代码
============================================================
📊 最终结果:
============================================================
会话ID: test_002
迭代次数: 1
错误状态: no
代码生成: ✓
代码调试: ✓
代码验证: ✓
代码执行: ✓
✅ 所有节点测试通过
最简单的测试脚本,无需任何外部依赖!
- ✅ 无需 Redis
- ✅ 无需 RabbitMQ
- ✅ 无需真实 LLM 服务
- ✅ 无需复杂的依赖库(pika, redis 等)
- ✅ 纯 Python 实现
使用方法:
python test_simple.py测试内容:
- 单个 Coder 节点测试
- 所有节点顺序执行测试
- 完整工作流模拟测试
完整的集成测试套件,包含更多测试场景和 pytest 支持。
使用方法:
# 方式 1: 直接运行
python test_main_integration.py
# 方式 2: 使用 pytest(需要安装 pytest)
pip install pytest pytest-asyncio
pytest test_main_integration.py -v -s测试运行脚本,支持多种测试模式。
使用方法:
# Mock 模式(默认,推荐)
python run_tests.py --mode mock
# 演示模式
python run_tests.py --mode demo
# PyTest 模式
python run_tests.py --mode pytest
# 端到端模式(需要所有服务运行)
python run_tests.py --mode e2e┌─────────────────────────────────────────┐
│ LangGraph Workflow │
│ (langgraph_message_queue_nodes.py) │
└────────────┬────────────────────────────┘
│
┌─────────┼─────────┬──────────┐
│ │ │ │
┌──▼──┐ ┌──▼──┐ ┌──▼──┐ ┌────▼────┐
│Coder│ │Debug│ │Check│ │ Tool │
│ 生成 │ │ 调试 │ │ 验证 │ │ 执行 │
└──┬──┘ └──┬──┘ └──┬──┘ └────┬────┘
│ │ │ │
└────────┴────────┴──────────┘
│
┌─────────▼─────────┐
│ Message Queue │
│ (Mock 或 │
│ Redis/RabbitMQ) │
└───────────────────┘
-
Coder Node 🔨
- 功能:根据用户需求生成代码
- 输入:用户问题、上下文
- 输出:代码解决方案(prefix, imports, code)
-
Debugger Node 🐛
- 功能:调试和修复代码错误
- 输入:生成的代码
- 输出:调试结果、错误信息
-
Checker Node ✅
- 功能:验证代码是否符合用户需求
- 输入:代码 + 原始需求
- 输出:验证结果(is_valid, errors, requirements_met)
-
Tool Node ⚙️
- 功能:执行代码并返回结果
- 输入:代码 + 导入
- 输出:执行结果(success, output, error)
用户需求
↓
1. Coder: 生成代码
↓
2. Debugger: 调试代码
↓
3. Checker: 验证需求
↓
4. Tool: 执行代码
↓
返回结果
模拟消息队列客户端,自动响应请求:
client = SimpleMockQueueClient()
# 自动处理所有 coder, debugger, checker, tool 请求功能:
- ✓ 自动发布/订阅
- ✓ 模拟异步通信
- ✓ 生成合理的响应数据
{
"prefix": "生成的代码解决方案",
"imports": "# 必要的导入",
"code": "def example_function():\n return 'Hello!'"
}{
"has_errors": False,
"error_info": "",
"message": "代码无错误"
}{
"is_valid": True,
"message": "代码符合要求",
"errors": [],
"requirements_met": ["所有需求已满足"]
}{
"success": True,
"output": "代码执行成功",
"error": None
}验证单个 Coder 节点能否正常工作:
# 创建初始状态
state = {
"messages": [("user", "创建一个计算器函数")],
"session_id": "test_001",
...
}
# 执行 Coder
result = await coder.generate_code(state)
# 验证结果
assert result['error'] == 'no'
assert result['code_solution']['code']验证所有节点能否按顺序协同工作:
# 1. Coder 生成
state = await coder.generate_code(state)
# 2. Debugger 调试
state = await debugger.debug_code(state)
# 3. Checker 验证
state = await checker.check_code(state)
# 4. Tool 执行
state = await tool.execute_code(state)
# 验证整体流程
assert state['code_solution']
assert state['validation_result']['is_valid']验证不同类型的需求都能正确处理:
test_cases = [
"创建一个排序函数",
"实现一个 RAG 链",
"写一个计算斐波那契数列的函数"
]
for question in test_cases:
result = await coder.generate_code(state)
assert result['code_solution']修改 test_simple.py 中的响应生成方法:
def _mock_coder_response(self, request_message):
"""自定义 Coder 响应"""
return {
"success": True,
"data": {
"prefix": "我的自定义代码",
"imports": "import custom_module",
"code": "def my_function():\n pass"
}
}async def test_my_custom_case():
"""我的自定义测试"""
client = SimpleMockQueueClient()
coder = SimpleCoderNode(client)
state = SimpleGraphState({
"messages": [("user", "我的自定义需求")],
"session_id": "custom_test",
...
})
result = await coder.generate_code(state)
# 自定义断言
assert result['code_solution']['code']
print("✅ 自定义测试通过")async def test_error_handling():
"""测试错误处理"""
# 模拟失败场景
client = SimpleMockQueueClient()
# 修改 mock 返回错误响应
def mock_error_response(request_message):
return {
"success": False,
"error": "模拟的错误"
}
# 测试节点如何处理错误
...ModuleNotFoundError: No module named 'xxx'
解决: test_simple.py 不需要任何外部依赖!确保使用这个文件测试。
RuntimeError: no running event loop
解决: 使用 asyncio.run() 或 await 关键字:
asyncio.run(test_function())TimeoutError: 等待响应超时
解决: 检查 mock 响应是否正确触发,或增加超时时间。
- ✅ 使用
test_simple.py快速验证 - ✅ 修改代码后立即测试
- ✅ 添加新功能时同时添加测试
- ✅ 运行完整测试套件
python run_tests.py --mode pytest - ✅ 检查所有测试通过
- ✅ 查看测试覆盖率
- ✅ 启动所有真实服务
- ✅ 运行端到端测试
python run_tests.py --mode e2e - ✅ 验证生产环境配置
- 🚀 快速测试:
python test_simple.py - 🎭 无需依赖: 完全使用 Mock 组件
- ✅ 全面覆盖: 测试所有核心功能
- 📊 清晰输出: 详细的测试结果和日志
- 🔧 易于扩展: 简单添加新测试用例
开始测试你的代码助手吧!🎊