不需要任何外部服务(Redis, LLM服务器等),使用 mock 模拟所有响应:
# 直接运行 mock 测试
python run_tests.py --mode mock
# 或者运行演示
python run_tests.py --mode demo# 安装 pytest(如果未安装)
pip install pytest pytest-asyncio
# 运行 pytest 测试套件
python run_tests.py --mode pytest
# 或直接使用 pytest
pytest test_main_integration.py -v -s优点:
- ✅ 不需要任何外部服务
- ✅ 快速执行
- ✅ 可靠且可重复
- ✅ 适合开发和 CI/CD
运行:
python run_tests.py --mode mock测试内容:
- ✓ 单个 Coder 节点
- ✓ 所有节点(Coder, Debugger, Checker, Tool)
- ✓ 完整工作流
快速演示各个节点的功能:
python run_tests.py --mode demo使用 pytest 框架运行结构化测试:
python run_tests.py --mode pytest使用真实服务进行完整测试(需要所有服务运行):
python run_tests.py --mode e2e测试套件包含以下 mock 组件:
-
MockMessageQueueClient - 模拟 Redis/RabbitMQ
- 自动响应请求
- 模拟各个服务的响应
- 无需真实消息队列
-
MockSystemMonitor - 模拟监控系统
- 记录请求
- 健康检查
- 生成监控数据
-
模拟的 LLM 响应
- Coder: 根据问题生成不同的代码
- Debugger: 模拟调试结果
- Checker: 模拟验证结果
- Tool: 模拟代码执行
问题中包含 "RAG" 或 "retrieval" 时:
{
"prefix": "A RAG chain implementation using LCEL",
"imports": "from langchain_core.prompts import ChatPromptTemplate\n...",
"code": "# Build RAG chain\nrag_chain = prompt | model | parser"
}问题中包含 "add" 或 "sum" 时:
{
"prefix": "Simple addition function",
"imports": "",
"code": "def add_numbers(a, b):\n return a + b"
}{
"has_errors": False,
"error_info": "",
"message": "No errors found in code"
}{
"is_valid": True,
"message": "Code meets all requirements",
"errors": [],
"requirements_met": ["All requirements satisfied"],
"requirements_missing": []
}需要以下服务运行:
- Redis (localhost:6379)
# 使用 Docker 启动 Redis
docker run -d -p 6379:6379 redis:latest
# 或使用本地 Redis
redis-server- LLM 模拟服务器 (localhost:8000)
python simple_llm_server.py- Agent 服务
# Coder Service (localhost:8001)
python local_llm_service_restful.py --service coder --port 8001 --llm-url http://localhost:8000
# Checker Service (localhost:8002)
python local_llm_service_restful.py --service checker --port 8002 --llm-url http://localhost:8000
# Debugger Service (localhost:8003)
python local_llm_service_restful.py --service debugger --port 8003 --llm-url http://localhost:8000- 运行端到端测试
python run_tests.py --mode e2e测试脚本会自动检查所有必需服务:
✅ Redis (localhost:6379) 正在运行
✅ LLM Server (localhost:8000) 正在运行
✅ Coder Service (localhost:8001) 正在运行
✅ Checker Service (localhost:8002) 正在运行
✅ Debugger Service (localhost:8003) 正在运行┌─────────────────────────────────────────┐
│ Main LangGraph Application │
│ (main_langgraph.py) │
└────────────────┬────────────────────────┘
│
┌──────────┴──────────┐
│ │
┌─────▼─────┐ ┌──────▼──────┐
│ Workflow │ │ Message │
│ Nodes │◄───────┤ Queue │
└─────┬─────┘ └──────┬──────┘
│ │
│ ┌──────▼──────┐
│ │ Services │
│ │ (Coder, │
│ │ Debugger, │
│ │ Checker, │
│ │ Tool) │
│ └─────────────┘
│
┌─────▼─────────────────────────────┐
│ Mock Components │
│ - MockMessageQueueClient │
│ - MockSystemMonitor │
│ - Mock LLM Responses │
└───────────────────────────────────┘
-
单节点测试
- 测试每个节点独立功能
- 验证输入输出格式
- 检查错误处理
-
多节点测试
- 测试节点间协作
- 验证状态传递
- 检查数据流
-
完整工作流测试
- 测试端到端流程
- 验证最终结果
- 检查性能指标
async def test_coder_node_generates_code():
state = GraphState(
messages=[("user", "How do I build a RAG chain?")],
...
)
result = await coder_node.generate_code(state)
assert result["code_solution"]["code"]async def test_debugger_node():
state = GraphState(
code_solution={"code": "def test(): pass"},
...
)
result = await debugger_node.debug_code(state)
assert "generation" in resultasync def test_checker_node():
state = GraphState(
messages=[("user", "Create calculator")],
code_solution={"code": "def calc(): pass"},
...
)
result = await checker_node.check_code(state)
assert result["validation_result"]["is_valid"]async def test_complete_workflow():
initial_state = GraphState(
messages=[("user", "Create add function")],
...
)
result = await workflow.ainvoke(initial_state)
assert result["code_solution"]🔬 ============================================================
🔬 开始运行集成测试套件
🔬 ============================================================
============================================================
测试 1: 单个 Coder 节点
============================================================
📊 结果:
- 错误状态: no
- 迭代次数: 1
- 代码前缀: A simple Python function
- 代码内容: def hello_world():...
✅ 单节点测试完成
============================================================
测试 2: 所有节点
============================================================
1️⃣ 测试 Coder 节点...
✓ 生成代码: A RAG chain implementation using LCEL
2️⃣ 测试 Debugger 节点...
✓ 调试结果: {'has_errors': False, ...}
3️⃣ 测试 Checker 节点...
✓ 验证结果: True
4️⃣ 测试 Tool 节点...
✓ 执行结果: {'success': True, ...}
✅ 所有节点测试完成
============================================================
测试 3: 完整工作流
============================================================
🚀 执行工作流...
📊 工作流结果:
- 会话ID: workflow_test
- 迭代次数: 1
- 错误状态: no
💻 生成的代码:
前缀: Sample Python code
导入:
代码:
def hello_world():
print('Hello, World!')
return 'Hello, World!'
✅ 工作流测试完成
🎉 ============================================================
🎉 所有测试通过!
🎉 ============================================================
错误:
ModuleNotFoundError: No module named 'langgraph'
解决:
pip install langgraph langchain-core错误:
redis.exceptions.ConnectionError
解决:
- 使用 mock 模式:
python run_tests.py --mode mock - 或启动 Redis:
redis-server或docker run -d -p 6379:6379 redis
错误:
ModuleNotFoundError: No module named 'pytest'
解决:
pip install pytest pytest-asyncio错误:
Address already in use: 8000
解决:
# 查找占用端口的进程
lsof -i :8000
# 终止进程
kill -9 <PID>pip install langgraph langchain-corepip install -r requirements.txtrequirements.txt 内容:
langgraph>=0.1.0
langchain-core>=0.1.0
redis>=4.0.0
pika>=1.3.0
fastapi>=0.104.0
uvicorn>=0.24.0
pydantic>=2.0.0
pytest>=7.4.0
pytest-asyncio>=0.21.0
aiohttp>=3.9.0
requests>=2.31.0
psutil>=5.9.0
prometheus-client>=0.19.0
- 使用 mock 模式快速迭代
python run_tests.py --mode mock- 验证单个功能
python run_tests.py --mode demo- 运行完整测试套件
python run_tests.py --mode pytest- 检查覆盖率(可选)
pytest --cov=. test_main_integration.py- 启动所有服务
- 运行端到端测试
python run_tests.py --mode e2e- Mock 测试是默认推荐的方式 - 快速、可靠、无需外部依赖
- 端到端测试需要所有服务 - 仅在部署前运行
- 测试日志保存在 logs/ 目录 - 便于调试
- 可以扩展 mock 响应 - 在
test_main_integration.py中修改
test_main_integration.py- 主测试文件run_tests.py- 测试运行脚本main_langgraph.py- 主应用langgraph_message_queue_nodes.py- 工作流节点local_llm_service_restful.py- 服务实现simple_llm_server.py- LLM 模拟服务器
- 优先使用 mock 模式开发和测试
- 端到端测试仅在必要时运行
- 查看日志文件获取详细信息
- 遇到问题先查看故障排查部分