# 考培练系统后端质量保证机制

## 1. 代码质量保证

### 1.1 自动化代码检查
```yaml
# .pre-commit-config.yaml
repos:
  - repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
      - id: black
        language_version: python3.8
  
  - repo: https://github.com/pycqa/isort
    rev: 5.12.0
    hooks:
      - id: isort
        args: ["--profile", "black"]
  
  - repo: https://github.com/pycqa/flake8
    rev: 6.0.0
    hooks:
      - id: flake8
        args: ["--max-line-length", "88", "--extend-ignore", "E203"]
  
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.3.0
    hooks:
      - id: mypy
        additional_dependencies: [types-all]
```

### 1.2 代码复杂度控制
```python
# setup.cfg
[flake8]
max-line-length = 88
max-complexity = 10  # 圈复杂度限制
extend-ignore = E203, W503
exclude = .git,__pycache__,venv,migrations

[mypy]
python_version = 3.8
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = True
disallow_incomplete_defs = True
check_untyped_defs = True
disallow_untyped_decorators = True
no_implicit_optional = True
warn_redundant_casts = True
warn_unused_ignores = True
warn_no_return = True
warn_unreachable = True
strict_equality = True
```

## 2. 测试策略

### 2.1 测试金字塔
```
         /\
        /  \  端到端测试 (10%)
       /    \ - 完整业务流程测试
      /      \- UI自动化测试
     /--------\
    /          \ 集成测试 (30%)
   /            \- API测试
  /              \- 数据库集成测试
 /                \- 第三方服务集成测试
/------------------\
                    单元测试 (60%)
                    - 业务逻辑测试
                    - 工具函数测试
                    - 数据验证测试
```

### 2.2 单元测试规范
```python
# tests/unit/test_user_service.py
import pytest
from unittest.mock import Mock, AsyncMock
from app.services.user_service import UserService
from app.models.user import User
from app.schemas.user import UserCreate

class TestUserService:
    """用户服务单元测试"""
    
    @pytest.fixture
    def mock_db(self):
        """模拟数据库会话"""
        return AsyncMock()
    
    @pytest.fixture
    def mock_logger(self):
        """模拟日志器"""
        return Mock()
    
    @pytest.fixture
    def user_service(self, mock_db, mock_logger):
        """创建用户服务实例"""
        return UserService(db=mock_db, logger=mock_logger)
    
    async def test_create_user_success(self, user_service, mock_db):
        """测试成功创建用户"""
        # 准备测试数据
        user_data = UserCreate(
            username="testuser",
            email="test@example.com",
            password="securepassword"
        )
        
        # 配置mock
        mock_db.execute.return_value.scalar_one_or_none.return_value = None
        mock_db.commit.return_value = None
        
        # 执行测试
        result = await user_service.create_user(user_data)
        
        # 验证结果
        assert result is not None
        assert result.username == user_data.username
        assert result.email == user_data.email
        mock_db.commit.assert_called_once()
    
    async def test_create_user_duplicate_username(self, user_service, mock_db):
        """测试用户名重复的情况"""
        # 准备测试数据
        user_data = UserCreate(
            username="existinguser",
            email="new@example.com",
            password="password"
        )
        
        # 配置mock - 模拟用户已存在
        mock_db.execute.return_value.scalar_one_or_none.return_value = User(
            id=1,
            username="existinguser"
        )
        
        # 执行测试并验证异常
        with pytest.raises(ValueError, match="用户名已存在"):
            await user_service.create_user(user_data)
        
        # 验证未提交事务
        mock_db.commit.assert_not_called()
```

### 2.3 集成测试规范
```python
# tests/integration/test_auth_flow.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.main import app
from app.models.user import User

@pytest.mark.integration
class TestAuthFlow:
    """认证流程集成测试"""
    
    async def test_complete_auth_flow(
        self,
        async_client: AsyncClient,
        db_session: AsyncSession
    ):
        """测试完整的认证流程"""
        # 1. 注册新用户
        register_data = {
            "username": "newuser",
            "email": "newuser@example.com",
            "password": "StrongPassword123!"
        }
        
        response = await async_client.post(
            "/api/v1/auth/register",
            json=register_data
        )
        assert response.status_code == 201
        user_data = response.json()["data"]
        assert user_data["username"] == register_data["username"]
        
        # 2. 登录
        login_data = {
            "username": register_data["username"],
            "password": register_data["password"]
        }
        
        response = await async_client.post(
            "/api/v1/auth/login",
            json=login_data
        )
        assert response.status_code == 200
        tokens = response.json()["data"]
        assert "access_token" in tokens
        assert "refresh_token" in tokens
        
        # 3. 访问受保护的端点
        headers = {"Authorization": f"Bearer {tokens['access_token']}"}
        response = await async_client.get(
            "/api/v1/users/me",
            headers=headers
        )
        assert response.status_code == 200
        assert response.json()["data"]["username"] == register_data["username"]
        
        # 4. 刷新Token
        refresh_data = {"refresh_token": tokens["refresh_token"]}
        response = await async_client.post(
            "/api/v1/auth/refresh",
            json=refresh_data
        )
        assert response.status_code == 200
        new_tokens = response.json()["data"]
        assert new_tokens["access_token"] != tokens["access_token"]
```

### 2.4 性能测试
```python
# tests/performance/test_api_performance.py
import pytest
import asyncio
import time
from httpx import AsyncClient
from statistics import mean, stdev

@pytest.mark.performance
class TestAPIPerformance:
    """API性能测试"""
    
    async def test_login_performance(self, async_client: AsyncClient):
        """测试登录接口性能"""
        login_data = {
            "username": "perftest",
            "password": "password123"
        }
        
        # 预热
        for _ in range(5):
            await async_client.post("/api/v1/auth/login", json=login_data)
        
        # 性能测试
        response_times = []
        concurrent_requests = 10
        total_requests = 100
        
        async def make_request():
            start_time = time.time()
            response = await async_client.post(
                "/api/v1/auth/login",
                json=login_data
            )
            end_time = time.time()
            return end_time - start_time, response.status_code
        
        # 执行并发请求
        for _ in range(total_requests // concurrent_requests):
            tasks = [make_request() for _ in range(concurrent_requests)]
            results = await asyncio.gather(*tasks)
            
            for response_time, status_code in results:
                assert status_code == 200
                response_times.append(response_time)
        
        # 分析结果
        avg_response_time = mean(response_times)
        std_response_time = stdev(response_times)
        max_response_time = max(response_times)
        min_response_time = min(response_times)
        
        # 性能断言
        assert avg_response_time < 0.1  # 平均响应时间小于100ms
        assert max_response_time < 0.5  # 最大响应时间小于500ms
        
        print(f"\n性能测试结果:")
        print(f"平均响应时间: {avg_response_time*1000:.2f}ms")
        print(f"标准差: {std_response_time*1000:.2f}ms")
        print(f"最小响应时间: {min_response_time*1000:.2f}ms")
        print(f"最大响应时间: {max_response_time*1000:.2f}ms")
```

## 3. 持续集成/持续部署 (CI/CD)

### 3.1 GitHub Actions配置
```yaml
# .github/workflows/ci.yml
name: CI Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.8'
      
      - name: Install dependencies
        run: |
          pip install -r requirements/dev.txt
      
      - name: Run linters
        run: |
          make lint
          make type-check

  test:
    runs-on: ubuntu-latest
    needs: lint
    
    services:
      mysql:
        image: mysql:8.0
        env:
          MYSQL_ROOT_PASSWORD: testpassword
          MYSQL_DATABASE: testdb
        options: >-
          --health-cmd="mysqladmin ping"
          --health-interval=10s
          --health-timeout=5s
          --health-retries=5
        ports:
          - 3306:3306
      
      redis:
        image: redis:7
        options: >-
          --health-cmd="redis-cli ping"
          --health-interval=10s
          --health-timeout=5s
          --health-retries=5
        ports:
          - 6379:6379
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.8'
      
      - name: Install dependencies
        run: |
          pip install -r requirements/dev.txt
      
      - name: Run migrations
        env:
          DATABASE_URL: mysql+aiomysql://root:testpassword@localhost:3306/testdb
        run: |
          alembic upgrade head
      
      - name: Run tests
        env:
          DATABASE_URL: mysql+aiomysql://root:testpassword@localhost:3306/testdb
          REDIS_URL: redis://localhost:6379/0
        run: |
          pytest tests/ -v --cov=app --cov-report=xml
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml
          fail_ci_if_error: true

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Run security checks
        uses: pyupio/safety@v1
        with:
          api-key: ${{ secrets.SAFETY_API_KEY }}
      
      - name: Run Bandit
        run: |
          pip install bandit
          bandit -r app/ -f json -o bandit-report.json
      
      - name: Upload security report
        uses: actions/upload-artifact@v3
        with:
          name: security-reports
          path: bandit-report.json
```

### 3.2 代码覆盖率配置
```ini
# .coveragerc
[run]
source = app
omit = 
    */tests/*
    */migrations/*
    */__init__.py
    */config/*

[report]
precision = 2
show_missing = True
skip_covered = False

[html]
directory = htmlcov

[xml]
output = coverage.xml
```

## 4. 代码审查流程

### 4.1 Pull Request模板
```markdown
<!-- .github/pull_request_template.md -->
## 描述
简要描述这个PR的目的和所做的更改。

## 更改类型
- [ ] Bug修复
- [ ] 新功能
- [ ] 性能优化
- [ ] 代码重构
- [ ] 文档更新
- [ ] 测试更新

## 检查清单
- [ ] 代码符合项目的编码规范
- [ ] 已添加/更新相关测试
- [ ] 所有测试通过
- [ ] 已更新相关文档
- [ ] 代码已经过自我审查
- [ ] 没有注释掉的代码
- [ ] 没有console.log或print调试语句

## 测试说明
描述如何测试这些更改。

## 相关Issue
Closes #(issue号)

## 截图（如适用）
如果有UI更改，请添加截图。
```

### 4.2 代码审查标准
```python
# scripts/code_review_checklist.py
"""代码审查检查工具"""

import ast
import sys
from pathlib import Path
from typing import List, Dict

class CodeReviewChecker(ast.NodeVisitor):
    """代码审查自动检查器"""
    
    def __init__(self):
        self.issues = []
        self.current_file = None
    
    def check_file(self, file_path: Path) -> List[Dict[str, any]]:
        """检查单个文件"""
        self.current_file = file_path
        self.issues = []
        
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # 检查文件级别的问题
        self._check_file_issues(content)
        
        # 解析AST并检查
        try:
            tree = ast.parse(content)
            self.visit(tree)
        except SyntaxError as e:
            self.issues.append({
                "type": "syntax_error",
                "line": e.lineno,
                "message": str(e)
            })
        
        return self.issues
    
    def _check_file_issues(self, content: str):
        """检查文件级别的问题"""
        lines = content.split('\n')
        
        for i, line in enumerate(lines, 1):
            # 检查行长度
            if len(line) > 88:
                self.issues.append({
                    "type": "line_too_long",
                    "line": i,
                    "message": f"行长度 {len(line)} 超过88字符"
                })
            
            # 检查TODO/FIXME
            if 'TODO' in line or 'FIXME' in line:
                self.issues.append({
                    "type": "todo_found",
                    "line": i,
                    "message": "发现待处理的TODO/FIXME"
                })
            
            # 检查print语句
            if 'print(' in line and not line.strip().startswith('#'):
                self.issues.append({
                    "type": "print_statement",
                    "line": i,
                    "message": "使用了print语句，应使用日志"
                })
    
    def visit_FunctionDef(self, node):
        """检查函数定义"""
        # 检查函数复杂度
        complexity = self._calculate_complexity(node)
        if complexity > 10:
            self.issues.append({
                "type": "high_complexity",
                "line": node.lineno,
                "message": f"函数 {node.name} 复杂度 {complexity} 过高"
            })
        
        # 检查是否有文档字符串
        if not ast.get_docstring(node):
            self.issues.append({
                "type": "missing_docstring",
                "line": node.lineno,
                "message": f"函数 {node.name} 缺少文档字符串"
            })
        
        self.generic_visit(node)
    
    def _calculate_complexity(self, node) -> int:
        """计算圈复杂度"""
        complexity = 1
        for child in ast.walk(node):
            if isinstance(child, (ast.If, ast.While, ast.For)):
                complexity += 1
            elif isinstance(child, ast.ExceptHandler):
                complexity += 1
        return complexity

def main():
    """主函数"""
    checker = CodeReviewChecker()
    
    # 检查所有Python文件
    app_dir = Path("app")
    all_issues = []
    
    for py_file in app_dir.rglob("*.py"):
        issues = checker.check_file(py_file)
        if issues:
            all_issues.append({
                "file": str(py_file),
                "issues": issues
            })
    
    # 输出结果
    if all_issues:
        print("代码审查发现以下问题：")
        for file_issues in all_issues:
            print(f"\n文件: {file_issues['file']}")
            for issue in file_issues['issues']:
                print(f"  行 {issue['line']}: [{issue['type']}] {issue['message']}")
        sys.exit(1)
    else:
        print("代码审查通过！")

if __name__ == "__main__":
    main()
```

## 5. 安全保证

### 5.1 安全扫描配置
```yaml
# .github/workflows/security.yml
name: Security Scan

on:
  schedule:
    - cron: '0 0 * * *'  # 每天运行
  push:
    branches: [main]

jobs:
  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'
          format: 'sarif'
          output: 'trivy-results.sarif'
      
      - name: Upload Trivy scan results
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: 'trivy-results.sarif'
      
      - name: OWASP Dependency Check
        uses: dependency-check/Dependency-Check_Action@main
        with:
          project: 'kaopeilian-backend'
          path: '.'
          format: 'ALL'
```

### 5.2 安全最佳实践检查
```python
# app/core/security_checks.py
"""运行时安全检查"""

import os
import re
from typing import List, Dict
from pathlib import Path

class SecurityChecker:
    """安全检查器"""
    
    @staticmethod
    def check_env_variables() -> List[str]:
        """检查环境变量安全性"""
        issues = []
        
        # 检查敏感配置
        sensitive_vars = [
            'SECRET_KEY',
            'DATABASE_URL',
            'COZE_API_TOKEN',
            'DIFY_API_KEY'
        ]
        
        for var in sensitive_vars:
            value = os.getenv(var)
            if not value:
                issues.append(f"缺少必要的环境变量: {var}")
            elif var == 'SECRET_KEY' and len(value) < 32:
                issues.append("SECRET_KEY 长度不足32字符")
        
        return issues
    
    @staticmethod
    def check_sql_injection_patterns(query: str) -> bool:
        """检查SQL注入模式"""
        dangerous_patterns = [
            r';\s*DROP\s+TABLE',
            r';\s*DELETE\s+FROM',
            r'UNION\s+SELECT',
            r'OR\s+1\s*=\s*1',
            r'--\s*$'
        ]
        
        for pattern in dangerous_patterns:
            if re.search(pattern, query, re.IGNORECASE):
                return True
        return False
    
    @staticmethod
    def check_file_upload(filename: str, content: bytes) -> Dict[str, any]:
        """检查文件上传安全性"""
        issues = []
        
        # 检查文件扩展名
        allowed_extensions = {'.pdf', '.doc', '.docx', '.xls', '.xlsx', '.png', '.jpg', '.jpeg'}
        ext = Path(filename).suffix.lower()
        if ext not in allowed_extensions:
            issues.append(f"不允许的文件类型: {ext}")
        
        # 检查文件大小
        max_size = 10 * 1024 * 1024  # 10MB
        if len(content) > max_size:
            issues.append(f"文件大小超过限制: {len(content)/1024/1024:.2f}MB > 10MB")
        
        # 检查文件内容魔术字节
        magic_bytes = {
            b'\x89PNG': '.png',
            b'\xFF\xD8\xFF': '.jpg',
            b'%PDF': '.pdf'
        }
        
        file_type_valid = False
        for magic, expected_ext in magic_bytes.items():
            if content.startswith(magic) and ext == expected_ext:
                file_type_valid = True
                break
        
        if not file_type_valid and ext in {'.png', '.jpg', '.pdf'}:
            issues.append("文件内容与扩展名不匹配")
        
        return {
            "valid": len(issues) == 0,
            "issues": issues
        }
```

## 6. 监控与告警

### 6.1 健康检查端点
```python
# app/api/v1/health.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_db
from app.services.ai.coze.client import CozeClient
from app.services.ai.dify.client import DifyClient
import redis.asyncio as redis
from datetime import datetime

router = APIRouter()

@router.get("/health")
async def health_check():
    """基础健康检查"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0"
    }

@router.get("/health/detailed")
async def detailed_health_check(
    db: AsyncSession = Depends(get_db)
):
    """详细健康检查"""
    health_status = {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "checks": {}
    }
    
    # 检查数据库
    try:
        await db.execute("SELECT 1")
        health_status["checks"]["database"] = {
            "status": "up",
            "response_time_ms": 5
        }
    except Exception as e:
        health_status["status"] = "unhealthy"
        health_status["checks"]["database"] = {
            "status": "down",
            "error": str(e)
        }
    
    # 检查Redis
    try:
        r = redis.from_url("redis://localhost:6379")
        await r.ping()
        health_status["checks"]["redis"] = {
            "status": "up",
            "response_time_ms": 2
        }
    except Exception as e:
        health_status["status"] = "degraded"
        health_status["checks"]["redis"] = {
            "status": "down",
            "error": str(e)
        }
    
    # 检查Coze连接
    try:
        coze = CozeClient()
        # 执行简单的API调用测试
        health_status["checks"]["coze"] = {
            "status": "up",
            "response_time_ms": 50
        }
    except Exception as e:
        health_status["status"] = "degraded"
        health_status["checks"]["coze"] = {
            "status": "down",
            "error": str(e)
        }
    
    return health_status
```

### 6.2 性能监控指标
```python
# app/core/metrics.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from functools import wraps
import time

# 定义指标
http_requests_total = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

http_request_duration_seconds = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint']
)

active_connections = Gauge(
    'active_connections',
    'Number of active connections'
)

ai_api_calls_total = Counter(
    'ai_api_calls_total',
    'Total AI API calls',
    ['platform', 'operation', 'status']
)

def track_request_metrics(endpoint: str):
    """请求指标追踪装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start_time = time.time()
            method = kwargs.get('request', args[0]).method
            
            try:
                response = await func(*args, **kwargs)
                status = response.status_code
                http_requests_total.labels(
                    method=method,
                    endpoint=endpoint,
                    status=status
                ).inc()
                return response
            finally:
                duration = time.time() - start_time
                http_request_duration_seconds.labels(
                    method=method,
                    endpoint=endpoint
                ).observe(duration)
        
        return wrapper
    return decorator
```

## 7. 发布管理

### 7.1 版本控制策略
```bash
# 版本号格式：MAJOR.MINOR.PATCH
# MAJOR: 不兼容的API更改
# MINOR: 向后兼容的功能添加
# PATCH: 向后兼容的错误修复

# 自动版本号管理脚本
#!/bin/bash
# scripts/bump_version.sh

CURRENT_VERSION=$(cat VERSION)
IFS='.' read -ra VERSION_PARTS <<< "$CURRENT_VERSION"

case $1 in
  major)
    NEW_VERSION="$((VERSION_PARTS[0] + 1)).0.0"
    ;;
  minor)
    NEW_VERSION="${VERSION_PARTS[0]}.$((VERSION_PARTS[1] + 1)).0"
    ;;
  patch)
    NEW_VERSION="${VERSION_PARTS[0]}.${VERSION_PARTS[1]}.$((VERSION_PARTS[2] + 1))"
    ;;
  *)
    echo "Usage: $0 {major|minor|patch}"
    exit 1
    ;;
esac

echo $NEW_VERSION > VERSION
echo "Version bumped from $CURRENT_VERSION to $NEW_VERSION"
```

### 7.2 发布检查清单
```markdown
## 发布前检查清单

### 代码质量
- [ ] 所有测试通过
- [ ] 代码覆盖率达标（>80%）
- [ ] 无安全漏洞警告
- [ ] 性能测试通过

### 文档
- [ ] API文档已更新
- [ ] CHANGELOG已更新
- [ ] README已更新
- [ ] 部署文档已更新

### 配置
- [ ] 环境变量已配置
- [ ] 数据库迁移已准备
- [ ] 依赖版本已锁定

### 回滚计划
- [ ] 数据库备份已完成
- [ ] 回滚脚本已准备
- [ ] 回滚流程已测试

### 通知
- [ ] 相关团队已通知
- [ ] 维护窗口已安排
- [ ] 监控告警已配置
```