init code
This commit is contained in:
7
app/__init__.py
Normal file
7
app/__init__.py
Normal file
@@ -0,0 +1,7 @@
|
||||
"""
|
||||
app/__init__.py - 应用包初始化文件
|
||||
"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "AI Agent Team"
|
||||
|
||||
305
app/agents.py
Normal file
305
app/agents.py
Normal file
@@ -0,0 +1,305 @@
|
||||
"""
|
||||
app/agents.py - 多Agent系统：PM Agent -> QA Agent -> Dev Agent 的工作流
|
||||
"""
|
||||
|
||||
import logging
|
||||
from app.config import get_settings
|
||||
import json
|
||||
from openai import OpenAI
|
||||
|
||||
from app.message import send_requirement_result, send_test_cases, send_message_to_feishu, send_generate_code
|
||||
from app.models import RequirementAnalysis, TestCaseResult, CodeGenerationResult
|
||||
|
||||
# 初始化日志
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PMAgent:
|
||||
"""产品经理Agent - 完善和扩展需求"""
|
||||
|
||||
def __init__(self):
|
||||
self.settings = get_settings()
|
||||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||||
|
||||
def analyze_requirement(self, simple_requirement: str) -> RequirementAnalysis:
|
||||
"""
|
||||
分析和完善简单的需求描述
|
||||
|
||||
Args:
|
||||
simple_requirement: 用户提供的简单需求描述
|
||||
|
||||
Returns:
|
||||
RequirementAnalysis: 包含完善后的需求信息
|
||||
"""
|
||||
|
||||
prompt = f"""你是一个资深的产品经理。请根据以下简单的需求描述，进行深入分析和完善。
|
||||
|
||||
需求描述：{simple_requirement}
|
||||
|
||||
请按以下格式返回JSON结果（必须是有效的JSON格式）：
|
||||
{{
|
||||
"functional_requirements": ["功能需求1", "功能需求2", ...],
|
||||
"non_functional_requirements": ["非功能需求1", "非功能需求2", ...],
|
||||
"acceptance_criteria": ["验收标准1", "验收标准2", ...],
|
||||
"edge_cases": ["边缘情况1", "边缘情况2", ...],
|
||||
"summary": "需求总结说明"
|
||||
}}
|
||||
|
||||
要求：
|
||||
1. 识别至少3个功能需求
|
||||
2. 识别至少2个非功能需求（性能、安全、可维护性等）
|
||||
3. 为每个功能列出至少2个严格的验收标准
|
||||
4. 识别至少3个潜在的边缘情况
|
||||
5. 提供清晰的需求总结
|
||||
|
||||
返回ONLY JSON内容，不要有其他文字。"""
|
||||
|
||||
response = self.client.chat.completions.create(
|
||||
model=self.settings.model,
|
||||
messages=[
|
||||
{"role": "user", "content": prompt}
|
||||
],
|
||||
temperature=0.7,
|
||||
max_tokens=2000
|
||||
)
|
||||
|
||||
# 提取响应内容
|
||||
content = response.choices[0].message.content
|
||||
|
||||
# 解析JSON
|
||||
try:
|
||||
result = json.loads(content)
|
||||
return RequirementAnalysis(**result)
|
||||
except json.JSONDecodeError:
|
||||
# 如果JSON解析失败，尝试提取JSON部分
|
||||
import re
|
||||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||||
if json_match:
|
||||
result = json.loads(json_match.group())
|
||||
return RequirementAnalysis(**result)
|
||||
raise ValueError(f"无法解析Agent响应: {content}")
|
||||
|
||||
|
||||
class QAAgent:
|
||||
"""质量保证Agent - 生成测试用例"""
|
||||
|
||||
def __init__(self):
|
||||
self.settings = get_settings()
|
||||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||||
|
||||
def generate_test_cases(self, requirement_analysis: RequirementAnalysis) -> TestCaseResult:
|
||||
"""
|
||||
基于需求分析生成测试用例
|
||||
|
||||
Args:
|
||||
requirement_analysis: PM Agent的分析结果
|
||||
|
||||
Returns:
|
||||
TestCaseResult: 包含测试用例的结果
|
||||
"""
|
||||
|
||||
# 构建需求信息
|
||||
requirement_text = f"""
|
||||
功能需求：
|
||||
{chr(10).join(f"- {req}" for req in requirement_analysis["functional_requirements"])}
|
||||
|
||||
非功能需求：
|
||||
{chr(10).join(f"- {req}" for req in requirement_analysis["non_functional_requirements"])}
|
||||
|
||||
验收标准：
|
||||
{chr(10).join(f"- {ac}" for ac in requirement_analysis["acceptance_criteria"])}
|
||||
|
||||
边缘情况：
|
||||
{chr(10).join(f"- {ec}" for ec in requirement_analysis["edge_cases"])}
|
||||
|
||||
需求总结：
|
||||
{requirement_analysis["summary"]}
|
||||
"""
|
||||
|
||||
prompt = f"""你是一个资深的QA工程师。根据以下需求信息，生成全面的测试用例和测试策略。
|
||||
|
||||
{requirement_text}
|
||||
|
||||
请返回JSON格式的结果（必须是有效的JSON格式）：
|
||||
{{
|
||||
"test_cases": [
|
||||
{{
|
||||
"test_id": "TC001",
|
||||
"test_name": "测试用例名称",
|
||||
"precondition": "前置条件",
|
||||
"steps": ["步骤1", "步骤2", ...],
|
||||
"expected_result": "预期结果",
|
||||
"test_type": "功能测试|性能测试|安全测试"
|
||||
}},
|
||||
...
|
||||
],
|
||||
"test_strategy": "测试策略说明",
|
||||
"coverage_plan": "测试覆盖计划"
|
||||
}}
|
||||
|
||||
要求：
|
||||
1. 为每个功能需求生成至少2个测试用例
|
||||
2. 为每个边缘情况生成1个测试用例
|
||||
3. 生成至少1个性能测试用例
|
||||
4. 生成至少1个安全测试用例
|
||||
5. 测试用例要包含明确的步骤和预期结果
|
||||
|
||||
返回ONLY JSON内容，不要有其他文字。"""
|
||||
|
||||
response = self.client.chat.completions.create(
|
||||
model=self.settings.model,
|
||||
messages=[
|
||||
{"role": "user", "content": prompt}
|
||||
],
|
||||
temperature=0.7,
|
||||
max_tokens=3000
|
||||
)
|
||||
|
||||
content = response.choices[0].message.content
|
||||
|
||||
try:
|
||||
result = json.loads(content)
|
||||
return TestCaseResult(**result)
|
||||
except json.JSONDecodeError:
|
||||
import re
|
||||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||||
if json_match:
|
||||
result = json.loads(json_match.group())
|
||||
return TestCaseResult(**result)
|
||||
raise ValueError(f"无法解析QA Agent响应: {content}")
|
||||
|
||||
|
||||
class DevAgent:
|
||||
"""开发Agent - 生成Java代码和单元测试"""
|
||||
|
||||
def __init__(self):
|
||||
self.settings = get_settings()
|
||||
self.client = OpenAI(api_key=self.settings.api_key, base_url=self.settings.base_url)
|
||||
|
||||
def generate_code(
|
||||
self,
|
||||
requirement_analysis: RequirementAnalysis,
|
||||
test_cases: TestCaseResult
|
||||
) -> CodeGenerationResult:
|
||||
"""
|
||||
生成Java实现代码和单元测试代码
|
||||
|
||||
Args:
|
||||
requirement_analysis: PM Agent的分析结果
|
||||
test_cases: QA Agent的测试用例
|
||||
|
||||
Returns:
|
||||
CodeGenerationResult: 包含Java代码和单元测试代码
|
||||
"""
|
||||
|
||||
# 构建上下文
|
||||
requirement_text = f"""
|
||||
功能需求：
|
||||
{chr(10).join(f"- {req}" for req in requirement_analysis["functional_requirements"])}
|
||||
|
||||
非功能需求：
|
||||
{chr(10).join(f"- {req}" for req in requirement_analysis["non_functional_requirements"])}
|
||||
|
||||
验收标准：
|
||||
{chr(10).join(f"- {ac}" for ac in requirement_analysis["acceptance_criteria"])}
|
||||
|
||||
需求总结：
|
||||
{requirement_analysis["summary"]}
|
||||
"""
|
||||
|
||||
test_summary = f"""
|
||||
测试用例数量: {len(test_cases["test_cases"])}
|
||||
测试策略: {test_cases["test_strategy"]}
|
||||
覆盖计划: {test_cases["coverage_plan"]}
|
||||
"""
|
||||
|
||||
prompt = f"""你是一个资深的Java开发工程师。根据以下需求和测试用例，生成高质量的Java实现代码和单元测试代码。
|
||||
|
||||
{requirement_text}
|
||||
|
||||
{test_summary}
|
||||
|
||||
请返回JSON格式的结果（必须是有效的JSON格式）：
|
||||
{{
|
||||
"java_code": "完整的Java实现代码（包含主类和必要的辅助类）",
|
||||
"unit_tests": "使用JUnit的单元测试代码",
|
||||
"implementation_notes": "实现说明和注意事项",
|
||||
"unit_tests_count": "生成的单元测试数量",
|
||||
"passed_tests_count": "通过的单元测试数量"
|
||||
}}
|
||||
|
||||
Java代码要求：
|
||||
1. 遵循Java编码规范和最佳实践
|
||||
2. 包含详细的代码注释
|
||||
3. 包含异常处理
|
||||
4. 支持所有的功能需求
|
||||
5. 考虑非功能需求（性能、安全等）
|
||||
|
||||
单元测试要求：
|
||||
1. 使用JUnit 4 或 JUnit 5
|
||||
2. 为每个公共方法生成测试
|
||||
3. 包含正常情况、边缘情况和异常情况的测试
|
||||
4. 使用有意义的测试方法名称
|
||||
5. 测试代码要清晰易读
|
||||
|
||||
通过的单元测试数量一定要是你真实运行了单元测试之后得到的真实数字，不能伪造
|
||||
|
||||
返回ONLY JSON内容，不要有其他文字。"""
|
||||
|
||||
response = self.client.chat.completions.create(
|
||||
model=self.settings.model,
|
||||
messages=[
|
||||
{"role": "user", "content": prompt}
|
||||
],
|
||||
temperature=0.7,
|
||||
max_tokens=8192
|
||||
)
|
||||
|
||||
content = response.choices[0].message.content
|
||||
|
||||
|
||||
try:
|
||||
result = json.loads(content)
|
||||
return CodeGenerationResult(**result)
|
||||
except json.JSONDecodeError:
|
||||
import re
|
||||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||||
if json_match:
|
||||
result = json.loads(json_match.group())
|
||||
return CodeGenerationResult(**result)
|
||||
raise ValueError(f"无法解析Dev Agent响应: {content}")
|
||||
|
||||
|
||||
async def orchestrate_agents(simple_requirement: str) -> dict:
|
||||
"""
|
||||
编排三个Agent的工作流程
|
||||
|
||||
Args:
|
||||
simple_requirement: 用户的简单需求描述
|
||||
|
||||
Returns:
|
||||
包含所有Agent结果的完整字典
|
||||
"""
|
||||
|
||||
send_message_to_feishu(f"收到新需求: {simple_requirement}")
|
||||
# Step 1: PM Agent 分析需求
|
||||
pm_agent = PMAgent()
|
||||
requirement_analysis = pm_agent.analyze_requirement(simple_requirement)
|
||||
send_requirement_result(requirement_analysis)
|
||||
|
||||
# Step 2: QA Agent 生成测试用例
|
||||
qa_agent = QAAgent()
|
||||
test_cases = qa_agent.generate_test_cases(requirement_analysis)
|
||||
send_test_cases(test_cases)
|
||||
|
||||
# Step 3: Dev Agent 生成代码
|
||||
dev_agent = DevAgent()
|
||||
code_generation = dev_agent.generate_code(requirement_analysis, test_cases)
|
||||
send_generate_code(code_generation)
|
||||
return {
|
||||
"requirement_analysis": requirement_analysis,
|
||||
"test_cases": test_cases,
|
||||
"code_generation": code_generation
|
||||
}
|
||||
|
||||
148
app/api.py
Normal file
148
app/api.py
Normal file
@@ -0,0 +1,148 @@
|
||||
"""
|
||||
app/api.py - FastAPI 应用主文件
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from pydantic import BaseModel
|
||||
|
||||
from app.agents import orchestrate_agents
|
||||
from app.config import get_settings
|
||||
|
||||
# 初始化日志
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 初始化设置
|
||||
settings = get_settings()
|
||||
|
||||
# 创建 FastAPI 应用
|
||||
app = FastAPI(
|
||||
title=settings.project_name,
|
||||
version=settings.project_version,
|
||||
debug=settings.fastapi_debug,
|
||||
)
|
||||
|
||||
# 添加 CORS 中间件
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
class RequirementRequest(BaseModel):
|
||||
"""需求分析请求模型"""
|
||||
requirement: str
|
||||
|
||||
|
||||
class RequirementAnalysisResponse(BaseModel):
|
||||
"""需求分析响应"""
|
||||
functional_requirements: List[str]
|
||||
non_functional_requirements: List[str]
|
||||
acceptance_criteria: List[str]
|
||||
edge_cases: List[str]
|
||||
summary: str
|
||||
|
||||
|
||||
class TestCase(BaseModel):
|
||||
"""测试用例"""
|
||||
test_id: str
|
||||
test_name: str
|
||||
precondition: str
|
||||
steps: List[str]
|
||||
expected_result: str
|
||||
test_type: str
|
||||
|
||||
|
||||
class TestCaseResponse(BaseModel):
|
||||
"""测试用例响应"""
|
||||
test_cases: List[TestCase]
|
||||
test_strategy: str
|
||||
coverage_plan: str
|
||||
|
||||
|
||||
class CodeGenerationResponse(BaseModel):
|
||||
"""代码生成响应"""
|
||||
java_code: str
|
||||
unit_tests: str
|
||||
implementation_notes: str
|
||||
|
||||
|
||||
class FullWorkflowResponse(BaseModel):
|
||||
"""完整工作流响应"""
|
||||
requirement_analysis: RequirementAnalysisResponse
|
||||
test_cases: TestCaseResponse
|
||||
code_generation: CodeGenerationResponse
|
||||
status: str = "success"
|
||||
|
||||
|
||||
# 路由
|
||||
|
||||
@app.post("/workflow/full", response_model=FullWorkflowResponse)
|
||||
async def full_workflow(request: RequirementRequest):
|
||||
"""
|
||||
完整工作流：PM Agent -> QA Agent -> Dev Agent
|
||||
|
||||
这个端点接收一个简单的需求描述，通过三个Agent的协作，
|
||||
最终生成完整的需求分析、测试用例和Java代码。
|
||||
|
||||
Args:
|
||||
request: RequirementRequest 包含简单的需求描述
|
||||
|
||||
Returns:
|
||||
FullWorkflowResponse 包含：
|
||||
- requirement_analysis: PM Agent的需求分析结果
|
||||
- test_cases: QA Agent生成的测试用例
|
||||
- code_generation: Dev Agent生成的Java代码和单元测试
|
||||
|
||||
Example:
|
||||
{
|
||||
"requirement": "创建一个用户管理系统，支持用户注册、登录、个人信息管理"
|
||||
}
|
||||
"""
|
||||
try:
|
||||
logger.info(f"开始处理需求: {request.requirement}")
|
||||
|
||||
|
||||
# 调用编排函数，执行三个Agent的工作流
|
||||
result = await orchestrate_agents(request.requirement)
|
||||
|
||||
logger.info("需求处理完成，正在组装响应")
|
||||
|
||||
# 构建响应
|
||||
response = FullWorkflowResponse(
|
||||
requirement_analysis=result["requirement_analysis"],
|
||||
test_cases=result["test_cases"],
|
||||
code_generation=result["code_generation"],
|
||||
status="success"
|
||||
)
|
||||
|
||||
logger.info("响应组装完成")
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理需求时出错: {str(e)}", exc_info=True)
|
||||
return {
|
||||
"error": str(e),
|
||||
"status": "error"
|
||||
}
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def root():
|
||||
"""根端点"""
|
||||
return {
|
||||
"message": "欢迎使用 AI Agent API",
|
||||
"endpoints": {
|
||||
"health": "/health",
|
||||
"chat": "/chat",
|
||||
"workflow": "/workflow/full",
|
||||
"docs": "/docs",
|
||||
}
|
||||
}
|
||||
|
||||
38
app/config.py
Normal file
38
app/config.py
Normal file
@@ -0,0 +1,38 @@
|
||||
"""
|
||||
app/config.py - 应用配置管理
|
||||
"""
|
||||
|
||||
from pydantic_settings import BaseSettings
|
||||
from functools import lru_cache
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""应用配置"""
|
||||
|
||||
# API 配置
|
||||
api_key: str
|
||||
model: str = "gemini-3.1-flash"
|
||||
base_url: str = "https://generativelanguage.googleapis.com/v1beta/openai/"
|
||||
|
||||
# FastAPI 配置
|
||||
fastapi_host: str = "0.0.0.0"
|
||||
fastapi_port: int = 8000
|
||||
fastapi_debug: bool = True
|
||||
|
||||
# LangGraph 配置
|
||||
langgraph_debug: bool = False
|
||||
|
||||
# 项目信息
|
||||
project_name: str = "AI Agent Project"
|
||||
project_version: str = "0.1.0"
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
case_sensitive = False
|
||||
|
||||
|
||||
@lru_cache()
|
||||
def get_settings() -> Settings:
|
||||
"""获取应用配置的单例"""
|
||||
return Settings()
|
||||
|
||||
273
app/message.py
Normal file
273
app/message.py
Normal file
@@ -0,0 +1,273 @@
|
||||
from collections import defaultdict
|
||||
|
||||
import requests
|
||||
import json
|
||||
|
||||
from app.models import RequirementAnalysis, TestCaseResult, CodeGenerationResult
|
||||
|
||||
webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/0cd37406-31b8-424a-a3da-c11ce465ac29"
|
||||
|
||||
def send_message_to_feishu(message: str):
|
||||
data = {
|
||||
"msg_type": "text",
|
||||
"content": {
|
||||
"text": message
|
||||
}
|
||||
}
|
||||
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(data)
|
||||
)
|
||||
|
||||
def send_requirement_result(requirement_analysis: RequirementAnalysis):
|
||||
data = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {
|
||||
"wide_screen_mode": True
|
||||
},
|
||||
"header": {
|
||||
"title": {
|
||||
"tag": "plain_text",
|
||||
"content": "📌 需求分析结果"
|
||||
},
|
||||
"template": "blue"
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": f"---\n**📎 总结：**\n\n{requirement_analysis['summary']}"
|
||||
},
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": f"### 🧩 功能需求\n\n{chr(10).join([f'- {req}' for req in requirement_analysis['functional_requirements']])}"
|
||||
},
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": f"### 🚀 非功能需求\n\n{chr(10).join([f'- {req}' for req in requirement_analysis['non_functional_requirements']])}"
|
||||
},
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": f"### ⚠️ 边界情况\n\n{chr(10).join([f'- {case}' for case in requirement_analysis['edge_cases']])}"
|
||||
},
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": f"### ✅ 验收标准\n\n{chr(10).join([f'- {criteria}' for criteria in requirement_analysis['acceptance_criteria']])}"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(data)
|
||||
)
|
||||
|
||||
def send_generate_code(code_result: CodeGenerationResult):
|
||||
implementation_notes_data = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {
|
||||
"wide_screen_mode": True
|
||||
},
|
||||
"header": {
|
||||
"title": {
|
||||
"tag": "plain_text",
|
||||
"content": "📌 代码生成结果 - 实现思路"
|
||||
},
|
||||
"template": "green"
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": code_result['implementation_notes']
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
java_code_data = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {
|
||||
"wide_screen_mode": True
|
||||
},
|
||||
"header": {
|
||||
"title": {
|
||||
"tag": "plain_text",
|
||||
"content": "📌 代码生成结果 - 业务代码"
|
||||
},
|
||||
"template": "green"
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": code_result['java_code']
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
unit_tests_data = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {
|
||||
"wide_screen_mode": True
|
||||
},
|
||||
"header": {
|
||||
"title": {
|
||||
"tag": "plain_text",
|
||||
"content": "🚀 代码生成结果 - 测试代码"
|
||||
},
|
||||
"template": "green"
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": code_result['unit_tests']
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
passed_tests_data = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {
|
||||
"wide_screen_mode": True
|
||||
},
|
||||
"header": {
|
||||
"title": {
|
||||
"tag": "plain_text",
|
||||
"content": "🚀 代码生成结果 - 单元测试执行结果"
|
||||
},
|
||||
"template": "green"
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "markdown",
|
||||
"content": f"""
|
||||
### 📊 测试统计
|
||||
|
||||
- **总用例数：** {code_result['unit_tests_count']}
|
||||
- **通过：** {code_result['passed_tests_count']}
|
||||
"""
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(implementation_notes_data)
|
||||
)
|
||||
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(java_code_data)
|
||||
)
|
||||
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(unit_tests_data)
|
||||
)
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(passed_tests_data)
|
||||
)
|
||||
|
||||
def send_test_cases(test_case: TestCaseResult):
|
||||
|
||||
requests.post(
|
||||
webhook_url,
|
||||
headers={"Content-Type": "application/json"},
|
||||
data=json.dumps(build_full_feishu_card(test_case))
|
||||
)
|
||||
|
||||
|
||||
def build_full_feishu_card(data: TestCaseResult) -> dict:
|
||||
test_cases = data.get("test_cases", [])
|
||||
|
||||
type_counter = defaultdict(int)
|
||||
grouped_cases = defaultdict(list)
|
||||
|
||||
# 分组 + 统计
|
||||
for case in test_cases:
|
||||
t_type = case.get("test_type", "未分类")
|
||||
type_counter[t_type] += 1
|
||||
grouped_cases[t_type].append(case)
|
||||
|
||||
total = len(test_cases)
|
||||
|
||||
elements = []
|
||||
|
||||
# 📊 统计
|
||||
stats_text = "### 📊 测试统计\n\n"
|
||||
stats_text += f"- 总用例数：**{total}**\n"
|
||||
for k, v in type_counter.items():
|
||||
stats_text += f"- {k}：{v}\n"
|
||||
|
||||
elements.append({
|
||||
"tag": "markdown",
|
||||
"content": stats_text
|
||||
})
|
||||
|
||||
# 🗂 分组 + 全量展示
|
||||
for test_type, cases in grouped_cases.items():
|
||||
elements.append({
|
||||
"tag": "markdown",
|
||||
"content": f"---\n## 🗂 {test_type}"
|
||||
})
|
||||
|
||||
for case in cases:
|
||||
steps_text = "\n".join([f" {i+1}. {s}" for i, s in enumerate(case.get("steps", []))])
|
||||
|
||||
case_text = f"""### {case.get('test_id')} - {case.get('test_name')}
|
||||
|
||||
- **前置条件：** {case.get('precondition')}
|
||||
- **执行步骤：**
|
||||
{steps_text}
|
||||
- **期望结果：** {case.get('expected_result')}
|
||||
"""
|
||||
|
||||
elements.append({
|
||||
"tag": "markdown",
|
||||
"content": case_text
|
||||
})
|
||||
|
||||
# 🧠 测试策略
|
||||
elements.append({
|
||||
"tag": "markdown",
|
||||
"content": f"---\n### 🧠 测试策略\n\n{data.get('test_strategy')}"
|
||||
})
|
||||
|
||||
# 🎯 覆盖计划
|
||||
elements.append({
|
||||
"tag": "markdown",
|
||||
"content": f"---\n### 🎯 覆盖计划\n\n{data.get('coverage_plan')}"
|
||||
})
|
||||
|
||||
return {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {
|
||||
"wide_screen_mode": True
|
||||
},
|
||||
"header": {
|
||||
"title": {
|
||||
"tag": "plain_text",
|
||||
"content": "🧪 自动化测试用例结果"
|
||||
},
|
||||
"template": "blue"
|
||||
},
|
||||
"elements": elements
|
||||
}
|
||||
}
|
||||
|
||||
26
app/models.py
Normal file
26
app/models.py
Normal file
@@ -0,0 +1,26 @@
|
||||
from typing import TypedDict, List
|
||||
|
||||
|
||||
class RequirementAnalysis(TypedDict):
|
||||
"""需求分析结果"""
|
||||
functional_requirements: List[str]
|
||||
non_functional_requirements: List[str]
|
||||
acceptance_criteria: List[str]
|
||||
edge_cases: List[str]
|
||||
summary: str
|
||||
|
||||
|
||||
class TestCaseResult(TypedDict):
|
||||
"""测试用例结果"""
|
||||
test_cases: List[dict]
|
||||
test_strategy: str
|
||||
coverage_plan: str
|
||||
|
||||
|
||||
class CodeGenerationResult(TypedDict):
|
||||
"""代码生成结果"""
|
||||
java_code: str
|
||||
unit_tests: str
|
||||
implementation_notes: str
|
||||
unit_tests_count: str
|
||||
passed_tests_count: str
|
||||
Reference in New Issue
Block a user