🚀 基于 OpenClaw + Claude Code 的
端到端研发自动化系统培训课件

# 使用 AI 生成 Python 单元测试（pytest） $ claude-code "/tdd 为 users.py 生成完整的 pytest 单元测试，覆盖所有边界情况" # 生成的测试代码：test_users.py import pytest from fastapi.testclient import TestClient from ..main import app from ..database import TestingSessionLocal, engine, Base from ..models import User # 测试前创建数据库表 @pytest.fixture(scope="function") def db_session(): Base.metadata.create_all(bind=engine) db = TestingSessionLocal() yield db db.close() Base.metadata.drop_all(bind=engine) @pytest.fixture def client(db_session): def override_get_db(): try: yield db_session finally: pass app.dependency_overrides[get_db] = override_get_db with TestClient(app) as c: yield c app.dependency_overrides.clear() # 测试创建用户 def test_create_user_success(client): """测试成功创建用户""" user_data = { "username": "testuser", "email": "test@example.com", "password": "securepass123" } response = client.post("/users", json=user_data) assert response.status_code == 201 data = response.json() assert data["username"] == "testuser" assert data["email"] == "test@example.com" assert "id" in data assert "password" not in data # 密码不应返回 def test_create_user_duplicate_email(client, db_session): """测试重复邮箱创建失败""" # 先创建一个用户 existing_user = User(username="existing", email="dup@example.com") db_session.add(existing_user) db_session.commit() # 尝试用相同邮箱创建 user_data = { "username": "newuser", "email": "dup@example.com", "password": "securepass123" } response = client.post("/users", json=user_data) assert response.status_code == 400 assert "邮箱已被注册" in response.json()["detail"] def test_create_user_invalid_email(client): """测试无效邮箱格式""" user_data = { "username": "testuser", "email": "invalid-email", "password": "securepass123" } response = client.post("/users", json=user_data) assert response.status_code == 422 # 验证错误 def test_create_user_short_password(client): """测试密码长度不足""" user_data = { "username": "testuser", "email": "test@example.com", "password": "short" # 少于 8 位 } response = client.post("/users", json=user_data) assert response.status_code == 422 def test_get_users_pagination(client, db_session): """测试用户列表分页""" # 创建 25 个测试用户 for i in range(25): user = User(username=f"user{i}", email=f"user{i}@example.com") db_session.add(user) db_session.commit() # 获取第一页（20 条） response = client.get("/users?page=1&limit=20") assert response.status_code == 200 data = response.json() assert len(data["data"]) == 20 assert data["total"] == 25 # 获取第二页（剩余 5 条） response = client.get("/users?page=2&limit=20") data = response.json() assert len(data["data"]) == 5 def test_get_user_not_found(client): """测试获取不存在的用户""" response = client.get("/users/99999") assert response.status_code == 404 assert "用户不存在" in response.json()["detail"]

# 集成测试：用户下单完整流程 import pytest from httpx import AsyncClient from ..services.order_service import create_order from ..services.payment_service import process_payment @pytest.mark.asyncio async def test_complete_order_flow(): """测试完整的下单流程：创建订单 -> 扣减库存 -> 支付 -> 更新订单状态""" async with AsyncClient() as client: # Step 1: 创建用户 user_response = await client.post("/users", json={ "username": "order_test_user", "email": "order@test.com", "password": "testpass123" }) assert user_response.status_code == 201 user_id = user_response.json()["id"] # Step 2: 创建商品 product_response = await client.post("/products", json={ "name": "测试商品", "price": 99.99, "stock": 100 }) assert product_response.status_code == 201 product_id = product_response.json()["id"] # Step 3: 创建订单 order_response = await client.post("/orders", json={ "user_id": user_id, "items": [{ "product_id": product_id, "quantity": 2 }] }) assert order_response.status_code == 201 order_id = order_response.json()["id"] # Step 4: 验证库存扣减 product_detail = await client.get(f"/products/{product_id}") assert product_detail.json()["stock"] == 98 # 100 - 2 # Step 5: 处理支付 payment_response = await client.post(f"/orders/{order_id}/pay", json={ "payment_method": "alipay" }) assert payment_response.status_code == 200 # Step 6: 验证订单状态更新 order_detail = await client.get(f"/orders/{order_id}") assert order_detail.json()["status"] == "paid" # Step 7: 验证订单总金额 expected_total = 99.99 * 2 assert abs(order_detail.json()["total_amount"] - expected_total) < 0.01

# E2E 测试：用户注册到下单完整流程 import pytest from playwright.sync_api import Page, expect from datetime import datetime @pytest.mark.e2e def test_user_registration_to_order(page: Page, base_url: str): """E2E 测试：从用户注册到成功下单的完整流程""" # Step 1: 访问首页 page.goto(base_url) expect(page).to_have_title("在线商城") # Step 2: 点击注册按钮 page.click('text="注册"') expect(page).to_have_url(f"{base_url}/register") # Step 3: 填写注册表单 timestamp = datetime.now().strftime("%Y%m%d%H%M%S") page.fill('input[name="username"]', f"e2e_user_{timestamp}") page.fill('input[name="email"]', f"e2e_{timestamp}@test.com") page.fill('input[name="password"]', "TestPass123!") page.fill('input[name="confirm_password"]', "TestPass123!") # Step 4: 提交注册 page.click('button[type="submit"]') # Step 5: 验证注册成功并跳转 expect(page).to_have_url(f"{base_url}/login") expect(page.locator('.alert-success')).to_be_visible() # Step 6: 登录 page.fill('input[name="email"]', f"e2e_{timestamp}@test.com") page.fill('input[name="password"]', "TestPass123!") page.click('button[type="submit"]') # Step 7: 验证登录成功 expect(page).to_have_url(f"{base_url}/") expect(page.locator('.user-menu')).to_be_visible() # Step 8: 浏览商品 page.click('text="手机数码"') expect(page).to_have_url(f"{base_url}/products?category=electronics") # Step 9: 查看商品详情 page.click('.product-card:first-child') expect(page.locator('h1.product-name')).to_be_visible() # Step 10: 加入购物车 page.select_option('select[name="quantity"]', "2") page.click('button:has-text("加入购物车")') # Step 11: 验证购物车提示 expect(page.locator('.toast-success')).to_contain_text("已加入购物车") # Step 12: 进入购物车 page.click('.cart-icon') expect(page).to_have_url(f"{base_url}/cart") # Step 13: 验证购物车商品 expect(page.locator('.cart-item')).to_have_count(1) expect(page.locator('.cart-quantity')).to_contain_text("× 2") # Step 14: 结算 page.click('button:has-text("去结算")') expect(page).to_have_url(f"{base_url}/checkout") # Step 15: 填写收货地址 page.fill('input[name="receiver_name"]', "测试用户") page.fill('input[name="receiver_phone"]', "13800138000") page.fill('textarea[name="address"]', "北京市朝阳区测试路 1 号") # Step 16: 选择支付方式 page.click('input[value="alipay"]') # Step 17: 提交订单 page.click('button:has-text("提交订单")') # Step 18: 验证订单创建成功 expect(page).to_have_url(f"{base_url}/orders/\\d+") expect(page.locator('.order-status')).to_contain_text("待支付") # Step 19: 模拟支付 page.click('button:has-text("立即支付")') # Step 20: 验证支付成功 expect(page.locator('.payment-success')).to_be_visible() expect(page.locator('.order-status')).to_contain_text("已支付") # 截图保存 page.screenshot(path=f"screenshots/e2e_order_{timestamp}.png")

# UI 自动化测试：视觉回归测试 from selenium import webdriver from selenium.webdriver.common.by import By from PIL import Image import imagehash import pytest @pytest.mark.visual def test_homepage_visual_regression(): """首页视觉回归测试""" options = webdriver.ChromeOptions() options.add_argument('--headless') driver = webdriver.Chrome(options=options) try: # 访问首页 driver.get("https://example.com") driver.set_window_size(1920, 1080) # 截取当前屏幕 current_screenshot = "screenshots/current_homepage.png" driver.save_screenshot(current_screenshot) # 与基准截图对比 baseline_screenshot = "screenshots/baseline_homepage.png" hash_current = imagehash.average_hash(Image.open(current_screenshot)) hash_baseline = imagehash.average_hash(Image.open(baseline_screenshot)) # 计算差异 diff = hash_current - hash_baseline print(f"图片差异度：{diff}") # 差异阈值（超过 10 认为有明显变化） assert diff < 10, f"视觉回归测试失败：差异度过大 ({diff})" finally: driver.quit() @pytest.mark.accessibility def test_accessibility_compliance(): """无障碍访问测试（WCAG 2.1 AA）""" from axe_selenium_python import Axe driver = webdriver.Chrome() try: driver.get("https://example.com") # 运行无障碍检测 axe = Axe(driver) axe.inject() results = axe.run() # 断言无严重无障碍问题 assert len(results["violations"]) == 0, \ f"发现无障碍问题：{results['violations']}" finally: driver.quit()

// Jenkinsfile - 声明式流水线 pipeline { agent any environment { DOCKER_REGISTRY = 'registry.example.com' IMAGE_NAME = 'ecommerce-app' K8S_NAMESPACE = 'production' KUBECONFIG_CREDENTIALS_ID = 'k8s-prod-config' } tools { nodejs 'NodeJS 18' jdk 'JDK 17' } stages { stage('Checkout') { steps { checkout scm script { env.GIT_COMMIT_SHORT = sh(returnStdout: true, script: 'git rev-parse --short HEAD').trim() env.BUILD_VERSION = "${env.BUILD_NUMBER}-${env.GIT_COMMIT_SHORT}" } } } stage('Code Quality Check') { parallel { stage('Backend Lint') { steps { sh 'pip install flake8 black' sh 'flake8 backend/app/' sh 'black --check backend/app/' } } stage('Frontend Lint') { steps { dir('frontend') { sh 'npm ci' sh 'npm run lint' } } } stage('Security Scan') { steps { sh 'pip install bandit safety' sh 'bandit -r backend/app/' sh 'safety check' } } } } stage('Unit Tests') { steps { parallel { stage('Backend Tests') { steps { sh 'pip install pytest pytest-cov' sh 'pytest tests/unit/ --cov=./app --cov-report=xml' } post { always { junit 'tests/unit/results.xml' publishCoverage adapters: [coberturaAdapter('coverage.xml')], sourceFileEncoding: 'UTF-8' } } } stage('Frontend Tests') { steps { dir('frontend') { sh 'npm ci' sh 'npm run test:coverage' } } post { always { junit 'frontend/coverage/junit.xml' } } } } } } stage('Integration Tests') { steps { script { docker-compose -f docker-compose.test.yml up -d sleep 30 // 等待服务启动 } sh 'pytest tests/integration/ -v' } post { always { script { docker-compose -f docker-compose.test.yml down } } } } stage('Build Docker Image') { steps { script { env.DOCKER_IMAGE = "${DOCKER_REGISTRY}/${IMAGE_NAME}:${BUILD_VERSION}" } sh 'docker build -t ${DOCKER_IMAGE} -f backend/Dockerfile .' sh 'docker tag ${DOCKER_IMAGE} ${DOCKER_REGISTRY}/${IMAGE_NAME}:latest' } } stage('Push Docker Image') { steps { withCredentials([usernamePassword(credentialsId: 'docker-registry-creds', usernameVariable: 'DOCKER_USER', passwordVariable: 'DOCKER_PASS')]) { sh 'echo ${DOCKER_PASS} | docker login ${DOCKER_REGISTRY} -u ${DOCKER_USER} --password-stdin' sh 'docker push ${DOCKER_IMAGE}' sh 'docker push ${DOCKER_REGISTRY}/${IMAGE_NAME}:latest' } } } stage('Deploy to K8S') { when { branch 'main' } steps { withKubeConfig([credentialsId: '${KUBECONFIG_CREDENTIALS_ID}', serverUrl: '']) { sh 'kubectl set image deployment/${IMAGE_NAME} app=${DOCKER_IMAGE} -n ${K8S_NAMESPACE}' sh 'kubectl rollout status deployment/${IMAGE_NAME} -n ${K8S_NAMESPACE} --timeout=300s' } } } stage('Smoke Tests') { when { branch 'main' } steps { script { def healthCheck = sh( script: 'curl -s -o /dev/null -w "%{http_code}" https://api.example.com/health', returnStdout: true ).trim() if (healthCheck != '200') { error('健康检查失败！触发自动回滚...') } } } } } post { success { echo '🎉 构建和部署成功！' script { slackSend(color: 'good', message: "构建 #${env.BUILD_NUMBER} 成功！\n版本：${env.BUILD_VERSION}\n镜像：${env.DOCKER_IMAGE}") } } failure { echo '❌ 构建失败！' script { slackSend(color: 'danger', message: "构建 #${env.BUILD_NUMBER} 失败！\n请检查：${env.BUILD_URL}") } } always { cleanWs() } } }

# 后端服务 Dockerfile（多阶段构建优化） # Stage 1: 构建阶段 FROM python:3.12-slim AS builder WORKDIR /app # 安装构建依赖 RUN apt-get update && apt-get install -y \\ gcc \\ libpq-dev \\ && rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir --user -r requirements.txt # Stage 2: 运行阶段 FROM python:3.12-slim WORKDIR /app # 创建非 root 用户 RUN useradd -m -u 1000 appuser # 从构建阶段复制已安装的包 COPY --from=builder /root/.local /home/appuser/.local COPY --from=builder /app . # 复制应用代码 COPY backend/app/ ./app/ # 设置环境变量 ENV PATH=/home/appuser/.local/bin:$PATH \\ PYTHONUNBUFFERED=1 \\ PYTHONDONTWRITEBYTECODE=1 # 切换到非 root 用户 USER appuser # 暴露端口 EXPOSE 8000 # 健康检查 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \\ CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1 # 启动命令 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# 前端服务 Dockerfile（Nginx 托管静态资源） # Stage 1: 构建阶段 FROM node:18-alpine AS builder WORKDIR /app # 复制 package 文件 COPY package*.json ./ # 安装依赖 RUN npm ci # 复制源代码 COPY . . # 构建生产版本 RUN npm run build # Stage 2: Nginx 运行阶段 FROM nginx:alpine # 复制自定义 Nginx 配置 COPY nginx.conf /etc/nginx/conf.d/default.conf # 从构建阶段复制构建产物 COPY --from=builder /app/dist /usr/share/nginx/html # 暴露端口 EXPOSE 80 # 健康检查 HEALTHCHECK --interval=30s --timeout=3s \\ CMD wget --quiet --tries=1 --spider http://localhost/ || exit 1 # 启动 Nginx CMD ["nginx", "-g", "daemon off;"]

# docker-compose.yml - 本地开发环境 version: '3.8' services: backend: build: context: . dockerfile: backend/Dockerfile ports: - "8000:8000" environment: - DATABASE_URL=postgresql://postgres:postgres@db:5432/ecommerce - REDIS_URL=redis://redis:6379/0 - ENVIRONMENT=development volumes: - ./backend/app:/app/app depends_on: - db - redis networks: - ecommerce-network frontend: build: context: ./frontend dockerfile: Dockerfile.dev ports: - "3000:3000" environment: - REACT_APP_API_URL=http://localhost:8000 volumes: - ./frontend/src:/app/src depends_on: - backend networks: - ecommerce-network db: image: postgres:15-alpine environment: - POSTGRES_DB=ecommerce - POSTGRES_USER=postgres - POSTGRES_PASSWORD=postgres volumes: - postgres-data:/var/lib/postgresql/data ports: - "5432:5432" networks: - ecommerce-network redis: image: redis:7-alpine ports: - "6379:6379" networks: - ecommerce-network nginx: image: nginx:alpine ports: - "80:80" volumes: - ./nginx.conf:/etc/nginx/conf.d/default.conf depends_on: - backend - frontend networks: - ecommerce-network volumes: postgres-data: networks: ecommerce-network: driver: bridge

# k8s/deployment.yaml - 后端服务部署配置 apiVersion: apps/v1 kind: Deployment metadata: name: ecommerce-backend namespace: production labels: app: ecommerce component: backend spec: replicas: 3 selector: matchLabels: app: ecommerce component: backend strategy: type: RollingUpdate rollingUpdate: maxSurge: 1 maxUnavailable: 0 template: metadata: labels: app: ecommerce component: backend annotations: prometheus.io/scrape: "true" prometheus.io/port: "8000" spec: containers: - name: backend image: registry.example.com/ecommerce-app:v1.2.3 imagePullPolicy: Always ports: - containerPort: 8000 name: http protocol: TCP env: - name: DATABASE_URL valueFrom: secretKeyRef: name: db-credentials key: url - name: REDIS_URL valueFrom: configMapKeyRef: name: app-config key: redis-url - name: ENVIRONMENT value: "production" resources: requests: memory: "512Mi" cpu: "250m" limits: memory: "1Gi" cpu: "500m" livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 timeoutSeconds: 5 failureThreshold: 3 readinessProbe: httpGet: path: /ready port: 8000 initialDelaySeconds: 5 periodSeconds: 5 timeoutSeconds: 3 failureThreshold: 3 volumeMounts: - name: logs mountPath: /app/logs volumes: - name: logs emptyDir: {} affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: labelSelector: matchLabels: app: ecommerce component: backend topologyKey: kubernetes.io/hostname --- apiVersion: v1 kind: Service metadata: name: ecommerce-backend-service namespace: production spec: selector: app: ecommerce component: backend ports: - port: 80 targetPort: 8000 protocol: TCP name: http type: ClusterIP --- apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: ecommerce-backend-hpa namespace: production spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: ecommerce-backend minReplicas: 3 maxReplicas: 10 metrics: - type: Resource resource: name: cpu target: type: Utilization averageUtilization: 70 - type: Resource resource: name: memory target: type: Utilization averageUtilization: 80 behavior: scaleDown: stabilizationWindowSeconds: 300 policies: - type: Percent value: 10 periodSeconds: 60 scaleUp: stabilizationWindowSeconds: 0 policies: - type: Percent value: 100 periodSeconds: 15 - type: Pods value: 4 periodSeconds: 15 selectPolicy: Max

# k8s/ingress.yaml - Ingress 路由配置 apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: ecommerce-ingress namespace: production annotations: kubernetes.io/ingress.class: nginx cert-manager.io/cluster-issuer: letsencrypt-prod nginx.ingress.kubernetes.io/ssl-redirect: "true" nginx.ingress.kubernetes.io/proxy-body-size: "50m" nginx.ingress.kubernetes.io/rate-limit: "100" spec: tls: - hosts: - api.example.com - www.example.com secretName: ecommerce-tls rules: - host: api.example.com http: paths: - path: / pathType: Prefix backend: service: name: ecommerce-backend-service port: number: 80 - host: www.example.com http: paths: - path: / pathType: Prefix backend: service: name: ecommerce-frontend-service port: number: 80

# KubeSphere DevOps 流水线配置（jenkinsfile-kubesphere.yaml） apiVersion: devops.kubesphere.io/v1alpha3 kind: Pipeline metadata: name: ecommerce-pipeline namespace: ecommerce-project labels: app: ecommerce spec: pipeline: definition: cpsScmFlow: scm: type: git source: url: https://github.com/example/ecommerce.git credentialsId: github-credentials scriptPath: Jenkinsfile description: 电商系统 CI/CD 流水线 --- # KubeSphere 应用部署配置 apiVersion: app.k8s.io/v1beta1 kind: Application metadata: name: ecommerce-app namespace: ecommerce-project spec: selector: matchLabels: app: ecommerce componentKinds: - group: apps/v1 kind: Deployment - group: v1 kind: Service - group: networking.k8s.io/v1 kind: Ingress descriptor: version: "1.2.3" description: "电商系统应用" maintainers: - name: DevOps Team email: devops@example.com