基于 OpenClaw + Claude Code 的端到端研发自动化系统
从需求到部署的全流程自动化解决方案
随着 AI 技术的快速发展,传统软件研发流程正经历深刻变革。本方案旨在构建一个基于 OpenClaw 和 Claude Code 的端到端研发自动化系统,实现从需求分析到生产部署的全流程自动化,同时支持人机协同,提升研发效率和质量。
将产品需求自动转化为 PRD 文档和技术方案,减少人工翻译成本
基于技术方案自动生成前后端代码,支持多语言和多框架
自动生成单元测试、集成测试和 UI 自动化测试用例
CI/CD 流水线自动构建、测试和部署到 K8s 集群
关键节点支持人工审核和干预,确保质量和安全
基于反馈数据持续优化 AI 模型和研发流程
本技术方案模板适用于以下场景:
系统采用分层架构设计,包含基础设施层、平台层、服务层和应用层,各层之间通过标准化接口进行通信。
开源 AI Agent 执行框架,负责连接现实世界工具和 API
Anthropic 提供的 AI 编程助手,深度理解代码库
模型上下文协议,标准化 LLM 与外部工具的交互
负责任务分发、状态跟踪和资源管理
系统采用微服务架构,每个核心功能独立部署,通过 API Gateway 统一对外提供服务。
| 服务名称 | 职责 | 技术栈 | 端口 |
|---|---|---|---|
api-gateway |
API 网关、路由转发、认证鉴权 | Nginx + Lua / Kong | 80/443 |
requirement-service |
需求分析、PRD 生成 | Python FastAPI | 8001 |
design-service |
技术方案设计、架构评审 | Python FastAPI | 8002 |
codegen-service |
代码生成、代码审查 | Python FastAPI + Claude Code | 8003 |
test-service |
测试用例生成、测试执行 | Python + pytest + Playwright | 8004 |
deploy-service |
CI/CD 编排、K8s 部署 | Go + Kubernetes Client | 8005 |
monitor-service |
系统监控、日志聚合 | Prometheus + Grafana + ELK | 9090/3000 |
agent-orchestrator |
Agent 编排、任务调度 | Python + Celery + Redis | 8006 |
Python 3.12+
Java 17+ / Go 1.21+
PostgreSQL 15+
| 组件 | 技术选型 | 说明 |
|---|---|---|
| Agent 框架 | OpenClaw LangChain | OpenClaw 用于工具调用,LangChain 用于 Agent 编排 |
| LLM 引擎 | Claude Code Claude API | Claude Sonnet 4.6 / Opus 4.6,支持 GitHub Actions 集成 |
| 协议标准 | MCP | Model Context Protocol,标准化 LLM 工具调用 |
| 向量数据库 | Chroma Pinecone | 代码库语义搜索和知识检索 |
| Embedding | text-embedding-3-large | 代码和文档的向量化表示 |
| 领域 | 工具 | 用途 |
|---|---|---|
| 容器化 | Docker 24+, Docker Compose | 应用容器化和本地开发环境 |
| 容器编排 | Kubernetes 1.28+, KubeSphere 4.x | 生产环境容器编排和管理 |
| CI/CD | Jenkins 2.x, GitHub Actions, GitLab CI | 持续集成和持续部署 |
| 监控 | Prometheus + Grafana | 指标监控和可视化 |
| 日志 | ELK Stack (Elasticsearch, Logstash, Kibana) | 日志聚合和分析 |
| 链路追踪 | Jaeger / SkyWalking | 分布式链路追踪 |
| 配置管理 | ConfigMap, Secrets, Vault | 配置和敏感信息管理 |
| 字段名 | 类型 | 约束 | 说明 |
|---|---|---|---|
id |
BIGINT | PRIMARY KEY, AUTO_INCREMENT | 主键 ID |
project_code |
VARCHAR(50) | UNIQUE, NOT NULL | 项目编码 |
name |
VARCHAR(200) | NOT NULL | 项目名称 |
description |
TEXT | 项目描述 | |
status |
ENUM | DEFAULT 'active' | 状态:active, archived, deleted |
created_by |
BIGINT | FOREIGN KEY | 创建人 ID |
created_at |
TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间 |
updated_at |
TIMESTAMP | ON UPDATE CURRENT_TIMESTAMP | 更新时间 |
| 字段名 | 类型 | 约束 | 说明 |
|---|---|---|---|
id |
BIGINT | PRIMARY KEY | 主键 ID |
project_id |
BIGINT | FOREIGN KEY, NOT NULL | 所属项目 ID |
requirement_no |
VARCHAR(50) | UNIQUE | 需求编号 |
title |
VARCHAR(300) | NOT NULL | 需求标题 |
content |
LONGTEXT | 需求内容(原始输入) | |
prd_document |
LONGTEXT | PRD 文档(AI 生成) | |
priority |
ENUM | DEFAULT 'medium' | 优先级:high, medium, low |
status |
ENUM | DEFAULT 'draft' | 状态:draft, reviewed, approved, rejected, implemented |
ai_analysis |
JSON | AI 分析结果(结构化数据) | |
created_at |
TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间 |
| 字段名 | 类型 | 约束 | 说明 |
|---|---|---|---|
id |
BIGINT | PRIMARY KEY | 主键 ID |
requirement_id |
BIGINT | FOREIGN KEY, NOT NULL | 关联需求 ID |
version |
VARCHAR(20) | DEFAULT '1.0.0' | 版本号 |
architecture_design |
LONGTEXT | 架构设计文档 | |
database_design |
LONGTEXT | 数据库设计文档 | |
api_design |
LONGTEXT | API 接口设计文档 | |
tech_stack |
JSON | 技术栈配置 | |
review_status |
ENUM | DEFAULT 'pending' | 评审状态:pending, approved, rejected |
review_comments |
JSON | 评审意见 | |
created_at |
TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间 |
| 字段名 | 类型 | 约束 | 说明 |
|---|---|---|---|
id |
BIGINT | PRIMARY KEY | 主键 ID |
solution_id |
BIGINT | FOREIGN KEY | 技术方案 ID |
path |
VARCHAR(500) | NOT NULL | API 路径 |
method |
ENUM | NOT NULL | HTTP 方法:GET, POST, PUT, DELETE, PATCH |
summary |
VARCHAR(300) | API 简述 | |
request_schema |
JSON | 请求体 Schema (OpenAPI) | |
response_schema |
JSON | 响应体 Schema (OpenAPI) | |
auth_required |
BOOLEAN | DEFAULT true | 是否需要认证 |
rate_limit |
INT | DEFAULT 100 | 速率限制(次/分钟) |
generated_code |
LONGTEXT | 生成的代码 |
| 字段名 | 类型 | 约束 | 说明 |
|---|---|---|---|
id |
BIGINT | PRIMARY KEY | 主键 ID |
task_type |
ENUM | NOT NULL | 任务类型:requirement_analysis, design, codegen, test, deploy |
agent_id |
VARCHAR(100) | NOT NULL | 执行 Agent ID |
input_data |
JSON | 输入数据 | |
output_data |
JSON | 输出数据 | |
status |
ENUM | DEFAULT 'pending' | 状态:pending, running, completed, failed, cancelled |
progress |
INT | DEFAULT 0 | 进度百分比 (0-100) |
error_message |
TEXT | 错误信息 | |
started_at |
TIMESTAMP | 开始时间 | |
completed_at |
TIMESTAMP | 完成时间 | |
human_review_required |
BOOLEAN | DEFAULT false | 是否需要人工审核 |
human_review_status |
ENUM | DEFAULT null | 人工审核状态:pending, approved, rejected |
| 字段名 | 类型 | 约束 | 说明 |
|---|---|---|---|
id |
BIGINT | PRIMARY KEY | 主键 ID |
workflow_id |
VARCHAR(100) | NOT NULL | 工作流定义 ID |
project_id |
BIGINT | FOREIGN KEY | 项目 ID |
current_stage |
VARCHAR(50) | 当前阶段 | |
overall_status |
ENUM | DEFAULT 'running' | 整体状态 |
stage_results |
JSON | 各阶段结果 | |
created_at |
TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间 |
-- 常用查询索引
CREATE INDEX idx_requirements_project_status ON requirements(project_id, status);
CREATE INDEX idx_requirements_created_at ON requirements(created_at DESC);
CREATE INDEX idx_tasks_agent_status ON agent_tasks(agent_id, status);
CREATE INDEX idx_tasks_created_at ON agent_tasks(created_at DESC);
CREATE INDEX idx_api_solution_path ON api_definitions(solution_id, path);
-- 全文索引(用于搜索)
CREATE FULLTEXT INDEX idx_requirements_content ON requirements(content);
CREATE FULLTEXT INDEX idx_projects_name ON projects(name);
-- 复合索引优化
CREATE INDEX idx_workflow_project_stage ON workflow_instances(project_id, current_stage);
| Key 模式 | 数据类型 | TTL | 用途 |
|---|---|---|---|
project:{id} |
Hash | 1h | 项目基本信息缓存 |
requirement:{id} |
Hash | 30m | 需求详情缓存 |
task:{id}:status |
String | 24h | 任务状态实时查询 |
user:{id}:permissions |
Set | 2h | 用户权限缓存 |
api:rate_limit:{ip}:{endpoint} |
String | 1m | API 限流计数 |
session:{token} |
Hash | 24h | 用户会话 |
/api/v1/...{
"code": 200, // 状态码
"message": "success", // 消息
"data": {}, // 数据主体
"timestamp": 1710345600, // 时间戳
"trace_id": "abc123" // 链路追踪 ID
}
// 错误响应示例
{
"code": 400,
"message": "参数验证失败",
"data": {
"errors": [
{"field": "email", "message": "邮箱格式不正确"}
]
},
"timestamp": 1710345600,
"trace_id": "abc123"
}
/api/v1/requirements
创建需求
请求体:
{
"project_id": 1,
"title": "用户登录功能优化",
"content": "支持手机号验证码登录...",
"priority": "high"
}
响应:返回创建的需求对象(含 AI 分析的 PRD)
/api/v1/requirements/{id}
获取需求详情
响应:完整需求信息,包括 PRD 文档和 AI 分析结果
/api/v1/requirements?project_id=1&status=approved&page=1&size=20
需求列表查询
查询参数:project_id, status, priority, keyword, page, size
/api/v1/requirements/{id}/analyze
触发 AI 需求分析
响应:返回任务 ID,异步获取分析结果
/api/v1/solutions
创建技术方案
请求体:
{
"requirement_id": 1,
"tech_stack": {
"backend": "python_fastapi",
"database": "postgresql",
"frontend": "react"
}
}
/api/v1/solutions/{id}
获取技术方案详情
响应:包含架构设计、数据库设计、API 设计等完整内容
/api/v1/solutions/{id}/review
提交方案评审
请求体:
{
"action": "approve", // approve | reject
"comments": "方案合理,建议增加缓存层..."
}
/api/v1/solutions/{solution_id}/apis
获取方案的 API 列表
/api/v1/apis/generate
AI 生成 API 定义
请求体:技术方案 ID 或需求描述
响应:生成的 OpenAPI 规范文档
/api/v1/apis/export/openapi
导出 OpenAPI 规范
查询参数:solution_id, format (json|yaml)
/api/v1/tasks
创建 Agent 任务
请求体:
{
"task_type": "codegen",
"agent_id": "claude-code-agent",
"input_data": {
"solution_id": 1,
"module": "user_service"
},
"human_review_required": true
}
/api/v1/tasks/{id}
查询任务状态
响应:任务状态、进度、输出数据
/api/v1/tasks?agent_id=xxx&status=running
任务列表查询
/api/v1/tasks/{id}/review
人工审核任务结果
请求体:
{
"action": "approve", // approve | reject | request_changes
"comments": "代码质量良好,但需要补充单元测试"
}
/api/v1/tasks/{id}/cancel
取消任务
/api/v1/workflows/start
启动完整研发工作流
请求体:
{
"project_id": 1,
"requirement_ids": [1, 2, 3],
"workflow_template": "full_pipeline", // full_pipeline | codegen_only | test_only
"skip_stages": [] // 可选跳过的阶段
}
响应:工作流实例 ID
/api/v1/workflows/{instance_id}
查询工作流状态
响应:各阶段状态和结果
/api/v1/workflows/{instance_id}/pause
暂停工作流
/api/v1/workflows/{instance_id}/resume
恢复工作流
/api/v1/metrics/project/{id}/overview
项目概览统计
响应:需求数量、完成率、代码行数、测试覆盖率等
/api/v1/metrics/agent/performance
Agent 性能统计
查询参数:agent_id, start_date, end_date
// JWT Token 认证
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
// API Key 认证(服务端对服务端)
X-API-Key: your_api_key_here
// 权限范围
scopes:
- requirements:read, requirements:write
- solutions:read, solutions:write
- tasks:read, tasks:write, tasks:review
- workflows:execute
| 接口类型 | 限流值 | 说明 |
|---|---|---|
| 普通查询接口 | 100 次/分钟/IP | GET 请求 |
| 写入接口 | 30 次/分钟/IP | POST/PUT/DELETE |
| AI 任务触发接口 | 10 次/分钟/User | 消耗 AI 资源的接口 |
| 工作流启动接口 | 5 次/分钟/User | 高资源消耗操作 |
系统定义了多个专业化 Agent 角色,每个角色负责研发流程中的特定环节,通过协同工作完成端到端的自动化研发。
职责:将原始需求转化为结构化的 PRD 文档
使用工具:Claude Code, OpenClaw Skills, 需求模板库
输出产物:PRD 文档、用户故事、验收标准
职责:基于 PRD 设计完整的技术方案
使用工具:Claude Code, 架构模式库,设计模式库
输出产物:技术方案文档、架构图、数据库设计、API 规范
职责:根据技术方案生成后端代码
使用工具:Claude Code, OpenClaw, 代码模板库,Lint 工具
输出产物:源代码、单元测试、API 文档
职责:根据 API 规范生成前端代码
使用工具:Claude Code, React/Vue 模板,UI 组件库
输出产物:前端源代码、组件文档、E2E 测试
职责:生成和执行各类测试
使用工具:pytest, Playwright, Locust, Allure
输出产物:测试用例、测试脚本、测试报告
职责:负责 CI/CD 和部署
使用工具:Docker, Kubectl, Helm, Jenkins API, KubeSphere API
输出产物:部署配置、CI/CD 脚本、监控面板
职责:自动化代码审查
使用工具:SonarQube, ESLint, Pylint, Security Scanner
输出产物:代码审查报告、修复建议
职责:协调各 Agent,管理工作流
使用工具:任务调度器,通知服务,报表生成器
输出产物:项目计划、进度报告、风险清单
// Agent 间消息格式
{
"message_id": "uuid",
"from_agent": "requirement-analyst",
"to_agent": "solution-architect",
"message_type": "task_handoff",
"payload": {
"task_id": "task_123",
"input_artifacts": {
"prd_document": "s3://bucket/prd_123.md",
"user_stories": [...]
},
"context": {
"project_id": 1,
"priority": "high",
"deadline": "2026-03-20"
}
},
"timestamp": "2026-03-13T10:30:00Z"
}
// 使用 Redis Pub/Sub 或消息队列 (RabbitMQ/Kafka) 进行通信
# CLAUDE.md - Agent 配置文件
# 位置:项目根目录
# 项目规范
coding_standards:
language: python
framework: fastapi
style_guide: pep8
type_hints: required
# 代码审查规则
review_criteria:
- test_coverage_min: 80%
- security_scan: required
- performance_check: required
- documentation: required
# Agent 行为配置
agent_behavior:
auto_commit: false # 需要人工确认提交
human_review_points:
- database_schema_changes
- api_breaking_changes
- security_sensitive_code
# MCP 服务器配置
mcp_servers:
- name: github
command: npx
args: ["-y", "@modelcontextprotocol/server-github"]
- name: filesystem
command: npx
args: ["-y", "@modelcontextprotocol/server-filesystem"]
工作流状态转换:
pending → running (启动工作流)
running → stage_requirement_analysis (进入需求分析)
stage_requirement_analysis → stage_design (需求分析完成)
stage_design → stage_api_definition (设计完成)
stage_api_definition → stage_coding (API 定义完成)
stage_coding → stage_unit_test (编码完成)
stage_unit_test → stage_integration_test (单元测试通过)
stage_integration_test → stage_ci_cd (集成测试通过)
stage_ci_cd → stage_deployment (CI/CD 完成)
stage_deployment → stage_ui_acceptance (部署完成)
stage_ui_acceptance → completed (验收通过)
任意阶段 → failed (发生错误)
任意阶段 → paused (人工暂停)
paused → running (恢复执行)
failed → running (重试)
running → cancelled (取消)
// Jenkinsfile
pipeline {
agent any
environment {
DOCKER_REGISTRY = 'registry.example.com'
K8S_CLUSTER = 'production'
APP_NAME = 'myapp'
}
stages {
stage('Checkout') {
steps {
checkout scm
}
}
stage('Code Quality') {
steps {
sh 'pip install pylint black mypy'
sh 'pylint src/'
sh 'black --check src/'
sh 'mypy src/'
}
}
stage('Unit Test') {
steps {
sh 'pip install pytest pytest-cov'
sh 'pytest --cov=src --cov-report=xml tests/'
}
post {
always {
junit 'reports/*.xml'
publishCoverage adapters: [coberturaAdapter('coverage.xml')]
}
}
}
stage('Build Docker Image') {
steps {
script {
def version = sh(script: 'git describe --tags', returnStdout: true).trim()
sh "docker build -t ${DOCKER_REGISTRY}/${APP_NAME}:${version} ."
sh "docker tag ${DOCKER_REGISTRY}/${APP_NAME}:${version} ${DOCKER_REGISTRY}/${APP_NAME}:latest"
}
}
}
stage('Push to Registry') {
steps {
withCredentials([usernamePassword(credentialsId: 'docker-creds', usernameVariable: 'USER', passwordVariable: 'PASS')]) {
sh 'docker login -u $USER -p $PASS registry.example.com'
sh 'docker push registry.example.com/myapp:$VERSION'
}
}
}
stage('Deploy to K8s') {
steps {
sh 'kubectl set image deployment/${APP_NAME} app=${DOCKER_REGISTRY}/${APP_NAME}:${VERSION}'
sh 'kubectl rollout status deployment/${APP_NAME}'
}
}
stage('Smoke Test') {
steps {
sh 'python tests/smoke_test.py'
}
}
}
post {
success {
slackSend channel: '#deployments', color: 'good', message: "Deployment successful: ${APP_NAME} ${VERSION}"
}
failure {
slackSend channel: '#deployments', color: 'danger', message: "Deployment failed: ${APP_NAME} ${VERSION}"
}
}
}
# .github/workflows/deploy.yml
name: CI/CD Pipeline
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pytest pytest-cov pylint
- name: Lint
run: pylint src/
- name: Test
run: pytest --cov=src tests/
- name: Upload coverage
uses: codecov/codecov-action@v3
build:
needs: test
runs-on: ubuntu-latest
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: Build Docker image
run: docker build -t registry.example.com/app:${{ github.sha }} .
- name: Push to registry
uses: docker/build-push-action@v5
with:
push: true
tags: registry.example.com/app:${{ github.sha }}
deploy:
needs: build
runs-on: ubuntu-latest
environment: production
steps:
- name: Deploy to K8s
uses: azure/k8s-deploy@v4
with:
namespace: production
manifests: k8s/deployment.yaml
images: registry.example.com/app:${{ github.sha }}
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
postgresql-client \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 创建非 root 用户
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
# 暴露端口
EXPOSE 8000
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: backend-app
namespace: production
labels:
app: backend-app
spec:
replicas: 3
selector:
matchLabels:
app: backend-app
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
template:
metadata:
labels:
app: backend-app
spec:
containers:
- name: app
image: registry.example.com/backend-app:latest
ports:
- containerPort: 8000
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: db-secret
key: url
- name: REDIS_URL
valueFrom:
configMapKeyRef:
name: app-config
key: redis_url
resources:
requests:
memory: "256Mi"
cpu: "250m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: backend-app-service
namespace: production
spec:
selector:
app: backend-app
ports:
- port: 80
targetPort: 8000
type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: backend-app-ingress
namespace: production
annotations:
nginx.ingress.kubernetes.io/rate-limit: "100"
spec:
rules:
- host: api.example.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: backend-app-service
port:
number: 80
# 金丝雀发布配置 (Argo Rollouts)
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
name: backend-app
spec:
replicas: 10
strategy:
canary:
steps:
- setWeight: 10 # 10% 流量
- pause: {duration: 5m}
- setWeight: 25 # 25% 流量
- pause: {duration: 10m}
- setWeight: 50 # 50% 流量
- pause: {duration: 15m}
- setWeight: 100 # 全量
trafficRouting:
istio:
virtualService:
name: backend-app-vsvc
routes:
- primary
| 指标类型 | 具体指标 | 告警阈值 |
|---|---|---|
| 系统资源 | CPU 使用率、内存使用率、磁盘 IO | CPU > 80%, Memory > 85% |
| 应用性能 | QPS、响应时间 (P95/P99)、错误率 | P95 > 500ms, Error Rate > 1% |
| 业务指标 | 活跃用户数、订单量、转化率 | 同比下跌 > 20% |
| AI 任务 | 任务队列长度、平均执行时间、失败率 | Queue > 1000, Failure Rate > 5% |
# 结构化日志格式
{
"timestamp": "2026-03-13T10:30:00Z",
"level": "INFO",
"service": "requirement-service",
"trace_id": "abc123def456",
"span_id": "span789",
"message": "Requirement analysis completed",
"context": {
"requirement_id": 123,
"user_id": 456,
"duration_ms": 2345
}
}
# 日志级别使用规范:
# ERROR: 系统错误,需要立即处理
# WARNING: 异常情况,不影响核心功能
# INFO: 重要业务操作
# DEBUG: 调试信息(生产环境关闭)
| 告警级别 | 触发条件 | 通知渠道 | 响应时间 |
|---|---|---|---|
| P0 - 严重 | 服务不可用、数据丢失 | 电话 + 短信 + Slack | 5 分钟内 |
| P1 - 高 | 核心功能异常、错误率飙升 | 短信 + Slack | 15 分钟内 |
| P2 - 中 | 性能下降、非核心功能异常 | Slack + 邮件 | 1 小时内 |
| P3 - 低 | 预警信息、容量规划 | 邮件 | 24 小时内 |
# tests/test_user_service.py
import pytest
from src.services.user_service import UserService
from src.models.user import User
class TestUserService:
@pytest.fixture
def user_service(self, db_session):
return UserService(db_session)
async def test_create_user_success(self, user_service):
"""测试成功创建用户"""
user_data = {
"username": "testuser",
"email": "test@example.com",
"password": "SecurePass123!"
}
result = await user_service.create_user(user_data)
assert result is not None
assert result.username == "testuser"
assert result.email == "test@example.com"
assert hasattr(result, 'id')
async def test_create_user_duplicate_email(self, user_service, existing_user):
"""测试创建重复邮箱的用户"""
user_data = {
"username": "newuser",
"email": existing_user.email,
"password": "SecurePass123!"
}
with pytest.raises(ValueError, match="Email already exists"):
await user_service.create_user(user_data)
async def test_get_user_by_id(self, user_service, existing_user):
"""测试通过 ID 获取用户"""
user = await user_service.get_user_by_id(existing_user.id)
assert user is not None
assert user.id == existing_user.id
assert user.username == existing_user.username
async def test_delete_user(self, user_service, existing_user):
"""测试删除用户"""
success = await user_service.delete_user(existing_user.id)
assert success is True
deleted_user = await user_service.get_user_by_id(existing_user.id)
assert deleted_user is None
# 运行测试并生成覆盖率报告
# pytest tests/ --cov=src --cov-report=html --cov-fail-under=80
# tests/integration/test_api.py
import pytest
from httpx import AsyncClient
from src.main import app
@pytest.mark.asyncio
async def test_create_requirement_flow():
"""测试需求创建完整流程"""
async with AsyncClient(app=app, base_url="http://test") as ac:
# 1. 创建项目
project_resp = await ac.post("/api/v1/projects", json={
"name": "Test Project",
"description": "Integration test project"
})
assert project_resp.status_code == 201
project_id = project_resp.json()["data"]["id"]
# 2. 创建需求
req_resp = await ac.post("/api/v1/requirements", json={
"project_id": project_id,
"title": "Test Requirement",
"content": "This is a test requirement",
"priority": "high"
})
assert req_resp.status_code == 201
requirement_id = req_resp.json()["data"]["id"]
# 3. 触发 AI 分析
analyze_resp = await ac.post(f"/api/v1/requirements/{requirement_id}/analyze")
assert analyze_resp.status_code == 202
task_id = analyze_resp.json()["data"]["task_id"]
# 4. 轮询任务状态
for _ in range(30): # 最多等待 30 秒
task_resp = await ac.get(f"/api/v1/tasks/{task_id}")
status = task_resp.json()["data"]["status"]
if status == "completed":
break
elif status == "failed":
pytest.fail("Task failed")
await asyncio.sleep(1)
# 5. 验证 PRD 已生成
final_req = await ac.get(f"/api/v1/requirements/{requirement_id}")
prd = final_req.json()["data"]["prd_document"]
assert prd is not None
assert len(prd) > 0
# tests/e2e/test_login_flow.py
import pytest
from playwright.async_api import async_playwright
@pytest.mark.asyncio
async def test_user_login_and_dashboard():
"""测试用户登录和仪表盘访问"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
# 1. 访问登录页
await page.goto("https://app.example.com/login")
# 2. 填写登录表单
await page.fill('input[name="email"]', "test@example.com")
await page.fill('input[name="password"]', "SecurePass123!")
# 3. 提交登录
await page.click('button[type="submit"]')
# 4. 等待跳转
await page.wait_for_url("**/dashboard")
# 5. 验证仪表盘元素
await expect(page.locator("h1")).toContainText("Dashboard")
await expect(page.locator(".stats-card")).toHaveCount(4)
# 6. 截图对比(视觉回归测试)
await expect(page).toHaveScreenshot("dashboard.png")
await browser.close()
# tests/performance/locustfile.py
from locust import HttpUser, task, between
class APIUser(HttpUser):
wait_time = between(1, 3)
@task(3)
def get_requirement(self):
self.client.get("/api/v1/requirements/1")
@task(2)
def list_requirements(self):
self.client.get("/api/v1/requirements?page=1&size=20")
@task(1)
def create_requirement(self):
self.client.post("/api/v1/requirements", json={
"project_id": 1,
"title": "Load Test Requirement",
"content": "Testing performance",
"priority": "medium"
})
# 运行性能测试:
# locust -f tests/performance/locustfile.py --host=http://localhost:8000
| 指标 | 最低要求 | 目标值 |
|---|---|---|
| 单元测试覆盖率 | ≥ 80% | ≥ 90% |
| 集成测试通过率 | 100% | 100% |
| E2E 测试通过率 | ≥ 95% | 100% |
| 代码重复率 | ≤ 5% | ≤ 3% |
| 严重 Bug 数 | 0 | 0 |
| 性能回归 | ≤ 10% | ≤ 5% |
| 术语 | 解释 |
|---|---|
| OpenClaw | 开源 AI Agent 框架,专注于工具调用和现实世界连接 |
| Claude Code | Anthropic 推出的 AI 编程助手,深度集成 GitHub |
| MCP | Model Context Protocol,LLM 与外部工具交互的标准协议 |
| Agent | 能够自主感知、决策和执行的智能体 |
| KubeSphere | 基于 Kubernetes 的企业级容器平台 |
项目模板结构:
project-template/
├── backend/
│ ├── src/
│ │ ├── main.py
│ │ ├── config.py
│ │ ├── models/
│ │ ├── services/
│ │ ├── api/
│ │ └── utils/
│ ├── tests/
│ ├── requirements.txt
│ └── Dockerfile
├── frontend/
│ ├── src/
│ ├── public/
│ ├── package.json
│ └── Dockerfile
├── k8s/
│ ├── deployment.yaml
│ ├── service.yaml
│ └── ingress.yaml
├── .github/
│ └── workflows/
│ └── ci-cd.yml
├── docs/
│ ├── architecture.md
│ ├── api.md
│ └── deployment.md
├── CLAUDE.md
└── README.md
| 版本 | 日期 | 变更内容 | 作者 |
|---|---|---|---|
| v1.0.0 | 2026-03-13 | 初始版本发布 | AI Architecture Team |