测试与部署

第一章：测试基础

测试环境配置

# 安装测试依赖
pip install pytest pytest-asyncio httpx pytest-cov

# conftest.py
import pytest
import asyncio
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.main import app
from app.db.base import Base, get_db

# 测试数据库
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

@pytest.fixture(scope="session")
def event_loop():
    """创建事件循环"""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def test_engine():
    """创建测试数据库引擎"""
    engine = create_async_engine(TEST_DATABASE_URL, echo=True)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest.fixture
async def db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
    """创建测试数据库会话"""
    async_session = async_sessionmaker(
        test_engine,
        class_=AsyncSession,
        expire_on_commit=False
    )
    async with async_session() as session:
        yield session
        await session.rollback()

@pytest.fixture
async def client(db_session) -> AsyncGenerator[AsyncClient, None]:
    """创建测试客户端"""
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

    app.dependency_overrides.clear()

基础测试

# tests/test_api.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_root(client: AsyncClient):
    """测试根端点"""
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!"}

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    """测试创建用户"""
    user_data = {
        "username": "testuser",
        "email": "test@example.com",
        "password": "password123"
    }
    response = await 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  # 密码不应返回

@pytest.mark.asyncio
async def test_get_user(client: AsyncClient):
    """测试获取用户"""
    # 先创建用户
    user_data = {
        "username": "getuser",
        "email": "getuser@example.com",
        "password": "password123"
    }
    create_response = await client.post("/users/", json=user_data)
    user_id = create_response.json()["id"]

    # 获取用户
    response = await client.get(f"/users/{user_id}")
    assert response.status_code == 200
    assert response.json()["username"] == "getuser"

@pytest.mark.asyncio
async def test_get_user_not_found(client: AsyncClient):
    """测试获取不存在的用户"""
    response = await client.get("/users/99999")
    assert response.status_code == 404

参数化测试

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
@pytest.mark.parametrize("username,email,expected_status", [
    ("valid_user", "valid@example.com", 201),
    ("ab", "valid@example.com", 422),  # 用户名太短
    ("valid_user", "invalid-email", 422),  # 邮箱格式错误
    ("", "valid@example.com", 422),  # 空用户名
])
async def test_create_user_validation(
    client: AsyncClient,
    username: str,
    email: str,
    expected_status: int
):
    """测试用户创建验证"""
    user_data = {
        "username": username,
        "email": email,
        "password": "password123"
    }
    response = await client.post("/users/", json=user_data)
    assert response.status_code == expected_status

@pytest.mark.asyncio
@pytest.mark.parametrize("method,path,expected_status", [
    ("GET", "/", 200),
    ("GET", "/health", 200),
    ("GET", "/nonexistent", 404),
    ("POST", "/users/", 422),  # 缺少请求体
])
async def test_endpoints(
    client: AsyncClient,
    method: str,
    path: str,
    expected_status: int
):
    """测试多个端点"""
    response = await client.request(method, path)
    assert response.status_code == expected_status

第二章：高级测试

认证测试

import pytest
from httpx import AsyncClient

@pytest.fixture
async def auth_token(client: AsyncClient) -> str:
    """获取认证令牌"""
    # 创建测试用户
    await client.post("/users/", json={
        "username": "authuser",
        "email": "auth@example.com",
        "password": "password123"
    })

    # 登录获取令牌
    response = await client.post(
        "/token",
        data={
            "username": "authuser",
            "password": "password123"
        }
    )
    return response.json()["access_token"]

@pytest.fixture
def auth_headers(auth_token: str) -> dict:
    """认证请求头"""
    return {"Authorization": f"Bearer {auth_token}"}

@pytest.mark.asyncio
async def test_protected_endpoint(client: AsyncClient, auth_headers: dict):
    """测试受保护的端点"""
    response = await client.get("/users/me", headers=auth_headers)
    assert response.status_code == 200
    assert response.json()["username"] == "authuser"

@pytest.mark.asyncio
async def test_protected_endpoint_no_token(client: AsyncClient):
    """测试无令牌访问受保护端点"""
    response = await client.get("/users/me")
    assert response.status_code == 401

@pytest.mark.asyncio
async def test_protected_endpoint_invalid_token(client: AsyncClient):
    """测试无效令牌"""
    headers = {"Authorization": "Bearer invalid_token"}
    response = await client.get("/users/me", headers=headers)
    assert response.status_code == 401

依赖注入测试

import pytest
from unittest.mock import AsyncMock, patch
from httpx import AsyncClient
from app.main import app
from app.dependencies import get_external_service

@pytest.fixture
def mock_external_service():
    """模拟外部服务"""
    mock = AsyncMock()
    mock.fetch_data.return_value = {"external": "data"}
    return mock

@pytest.mark.asyncio
async def test_with_mocked_dependency(
    client: AsyncClient,
    mock_external_service
):
    """测试模拟依赖"""
    app.dependency_overrides[get_external_service] = lambda: mock_external_service

    response = await client.get("/external-data")
    assert response.status_code == 200
    assert response.json()["external"] == "data"

    # 验证调用
    mock_external_service.fetch_data.assert_called_once()

    app.dependency_overrides.clear()

# 使用 patch 模拟
@pytest.mark.asyncio
async def test_with_patch(client: AsyncClient):
    """使用 patch 模拟"""
    with patch("app.services.email.send_email") as mock_send:
        mock_send.return_value = True

        response = await client.post("/send-notification", json={
            "email": "test@example.com",
            "message": "Hello"
        })

        assert response.status_code == 200
        mock_send.assert_called_once_with(
            "test@example.com",
            "Hello"
        )

数据库测试

import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.crud.user import user_crud
from app.schemas.user import UserCreate

@pytest.mark.asyncio
async def test_create_user_db(db_session: AsyncSession):
    """测试数据库创建用户"""
    user_in = UserCreate(
        username="dbuser",
        email="db@example.com",
        password="password123"
    )
    user = await user_crud.create(db_session, obj_in=user_in)

    assert user.id is not None
    assert user.username == "dbuser"
    assert user.email == "db@example.com"

@pytest.mark.asyncio
async def test_get_user_by_email(db_session: AsyncSession):
    """测试根据邮箱获取用户"""
    # 创建用户
    user_in = UserCreate(
        username="emailuser",
        email="email@example.com",
        password="password123"
    )
    await user_crud.create(db_session, obj_in=user_in)

    # 查询
    user = await user_crud.get_by_email(db_session, email="email@example.com")
    assert user is not None
    assert user.username == "emailuser"

@pytest.mark.asyncio
async def test_update_user(db_session: AsyncSession):
    """测试更新用户"""
    # 创建用户
    user_in = UserCreate(
        username="updateuser",
        email="update@example.com",
        password="password123"
    )
    user = await user_crud.create(db_session, obj_in=user_in)

    # 更新
    updated_user = await user_crud.update(
        db_session,
        db_obj=user,
        obj_in={"full_name": "Updated Name"}
    )

    assert updated_user.full_name == "Updated Name"

集成测试

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_user_workflow(client: AsyncClient):
    """测试完整用户工作流"""
    # 1. 注册
    register_response = await client.post("/users/", json={
        "username": "workflow_user",
        "email": "workflow@example.com",
        "password": "password123"
    })
    assert register_response.status_code == 201
    user_id = register_response.json()["id"]

    # 2. 登录
    login_response = await client.post("/token", data={
        "username": "workflow_user",
        "password": "password123"
    })
    assert login_response.status_code == 200
    token = login_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    # 3. 获取个人信息
    me_response = await client.get("/users/me", headers=headers)
    assert me_response.status_code == 200
    assert me_response.json()["username"] == "workflow_user"

    # 4. 更新个人信息
    update_response = await client.put(
        "/users/me",
        headers=headers,
        json={"full_name": "Workflow User"}
    )
    assert update_response.status_code == 200
    assert update_response.json()["full_name"] == "Workflow User"

    # 5. 登出（如果实现了）
    # logout_response = await client.post("/logout", headers=headers)
    # assert logout_response.status_code == 200

第三章：Docker 部署

Dockerfile

# Dockerfile
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 设置环境变量
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1 \
    PIP_DISABLE_PIP_VERSION_CHECK=1

# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非 root 用户
RUN adduser --disabled-password --gecos "" appuser && \
    chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

多阶段构建

# Dockerfile.multistage
# 构建阶段
FROM python:3.11-slim as builder

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    libpq-dev

COPY requirements.txt .
RUN pip wheel --no-cache-dir --no-deps --wheel-dir /app/wheels -r requirements.txt

# 运行阶段
FROM python:3.11-slim

WORKDIR /app

# 安装运行时依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq5 \
    && rm -rf /var/lib/apt/lists/*

# 从构建阶段复制 wheels
COPY --from=builder /app/wheels /wheels
RUN pip install --no-cache /wheels/*

# 复制应用代码
COPY . .

# 创建非 root 用户
RUN adduser --disabled-password --gecos "" appuser && \
    chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose

# docker-compose.yml
version: "3.8"

services:
  app:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db:5432/fastapi
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    volumes:
      - ./app:/app/app # 开发模式热重载
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=fastapi
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./ssl:/etc/nginx/ssl:ro
    depends_on:
      - app

volumes:
  postgres_data:
  redis_data:

Nginx 配置

# nginx.conf
events {
    worker_connections 1024;
}

http {
    upstream fastapi {
        server app:8000;
        keepalive 32;
    }

    server {
        listen 80;
        server_name example.com;

        # 重定向到 HTTPS
        return 301 https://$server_name$request_uri;
    }

    server {
        listen 443 ssl http2;
        server_name example.com;

        ssl_certificate /etc/nginx/ssl/cert.pem;
        ssl_certificate_key /etc/nginx/ssl/key.pem;

        # SSL 配置
        ssl_protocols TLSv1.2 TLSv1.3;
        ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;
        ssl_prefer_server_ciphers off;

        # 安全头
        add_header X-Frame-Options "SAMEORIGIN" always;
        add_header X-Content-Type-Options "nosniff" always;
        add_header X-XSS-Protection "1; mode=block" always;

        location / {
            proxy_pass http://fastapi;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;

            # 超时设置
            proxy_connect_timeout 60s;
            proxy_send_timeout 60s;
            proxy_read_timeout 60s;
        }

        # 静态文件
        location /static {
            alias /app/static;
            expires 30d;
            add_header Cache-Control "public, immutable";
        }

        # 健康检查
        location /health {
            proxy_pass http://fastapi/health;
            access_log off;
        }
    }
}

第四章：生产部署

Gunicorn + Uvicorn

# gunicorn.conf.py
import multiprocessing

# 绑定
bind = "0.0.0.0:8000"

# Worker 配置
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 50

# 超时
timeout = 30
graceful_timeout = 30
keepalive = 5

# 日志
accesslog = "-"
errorlog = "-"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# 进程名
proc_name = "fastapi"

# 预加载应用
preload_app = True

# 钩子
def on_starting(server):
    print("Server is starting")

def on_exit(server):
    print("Server is shutting down")

def pre_fork(server, worker):
    print(f"Worker {worker.pid} is starting")

def post_fork(server, worker):
    print(f"Worker {worker.pid} has started")

# 启动命令
gunicorn app.main:app -c gunicorn.conf.py

系统服务

# /etc/systemd/system/fastapi.service
[Unit]
Description=FastAPI Application
After=network.target

[Service]
Type=notify
User=www-data
Group=www-data
WorkingDirectory=/var/www/fastapi
Environment="PATH=/var/www/fastapi/venv/bin"
Environment="PYTHONPATH=/var/www/fastapi"
ExecStart=/var/www/fastapi/venv/bin/gunicorn app.main:app -c gunicorn.conf.py
ExecReload=/bin/kill -s HUP $MAINPID
Restart=always
RestartSec=5

# 资源限制
LimitNOFILE=65535
LimitNPROC=4096

# 安全配置
NoNewPrivileges=yes
PrivateTmp=yes
ProtectSystem=strict
ProtectHome=yes
ReadWritePaths=/var/www/fastapi/logs

[Install]
WantedBy=multi-user.target

# 启用和启动服务
sudo systemctl daemon-reload
sudo systemctl enable fastapi
sudo systemctl start fastapi
sudo systemctl status fastapi

健康检查

from fastapi import FastAPI, Response
from datetime import datetime
import asyncio

app = FastAPI()

startup_time = None

@app.on_event("startup")
async def startup():
    global startup_time
    startup_time = datetime.now()

@app.get("/health")
async def health_check():
    """基础健康检查"""
    return {"status": "healthy"}

@app.get("/health/ready")
async def readiness_check():
    """就绪检查 - 检查所有依赖"""
    checks = {}

    # 检查数据库
    try:
        await check_database()
        checks["database"] = "healthy"
    except Exception as e:
        checks["database"] = f"unhealthy: {str(e)}"

    # 检查 Redis
    try:
        await check_redis()
        checks["redis"] = "healthy"
    except Exception as e:
        checks["redis"] = f"unhealthy: {str(e)}"

    # 检查外部服务
    try:
        await check_external_service()
        checks["external_service"] = "healthy"
    except Exception as e:
        checks["external_service"] = f"unhealthy: {str(e)}"

    all_healthy = all(v == "healthy" for v in checks.values())

    return Response(
        content={"status": "ready" if all_healthy else "not ready", "checks": checks},
        status_code=200 if all_healthy else 503
    )

@app.get("/health/live")
async def liveness_check():
    """存活检查"""
    return {
        "status": "alive",
        "uptime": str(datetime.now() - startup_time) if startup_time else None
    }

第五章：CI/CD 配置

GitHub Actions

# .github/workflows/ci.yml
name: CI/CD

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

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: test
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: "pip"

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio pytest-cov httpx

      - name: Run tests
        env:
          DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/test
        run: |
          pytest --cov=app --cov-report=xml --cov-report=html

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml

  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install linters
        run: pip install ruff mypy

      - name: Run Ruff
        run: ruff check .

      - name: Run MyPy
        run: mypy app

  build:
    needs: [test, lint]
    runs-on: ubuntu-latest
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Login to Container Registry
        uses: docker/login-action@v3
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Build and push
        uses: docker/build-push-action@v5
        with:
          context: .
          push: true
          tags: |
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

  deploy:
    needs: build
    runs-on: ubuntu-latest
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'

    steps:
      - name: Deploy to server
        uses: appleboy/ssh-action@v1.0.0
        with:
          host: ${{ secrets.SERVER_HOST }}
          username: ${{ secrets.SERVER_USER }}
          key: ${{ secrets.SERVER_SSH_KEY }}
          script: |
            cd /var/www/fastapi
            docker-compose pull
            docker-compose up -d
            docker system prune -f

环境配置

# app/core/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
from typing import Optional

class Settings(BaseSettings):
    # 应用配置
    APP_NAME: str = "FastAPI App"
    DEBUG: bool = False
    VERSION: str = "1.0.0"
    API_PREFIX: str = "/api/v1"

    # 数据库
    DATABASE_URL: str
    DB_POOL_SIZE: int = 20
    DB_MAX_OVERFLOW: int = 10

    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"

    # 安全
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7

    # CORS
    ALLOWED_ORIGINS: list[str] = ["*"]

    # 日志
    LOG_LEVEL: str = "INFO"

    class Config:
        env_file = ".env"
        case_sensitive = True

@lru_cache()
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

# .env.example
APP_NAME=FastAPI App
DEBUG=false

DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname
REDIS_URL=redis://localhost:6379/0

SECRET_KEY=your-super-secret-key-change-in-production

ALLOWED_ORIGINS=["https://example.com"]

LOG_LEVEL=INFO

常见问题

Q1：如何实现零停机部署？

# docker-compose.yml 使用滚动更新
services:
  app:
    deploy:
      replicas: 3
      update_config:
        parallelism: 1
        delay: 10s
        failure_action: rollback
      rollback_config:
        parallelism: 1
        delay: 10s

Q2：如何管理数据库迁移？

# CI/CD 中执行迁移
- name: Run migrations
  run: |
    docker-compose exec -T app alembic upgrade head

Q3：如何处理敏感配置？

# 使用 GitHub Secrets 或 HashiCorp Vault
- name: Create .env file
  run: |
    echo "DATABASE_URL=${{ secrets.DATABASE_URL }}" >> .env
    echo "SECRET_KEY=${{ secrets.SECRET_KEY }}" >> .env

学习资源

1. pytest 文档：https://docs.pytest.org/
2. Docker 最佳实践：https://docs.docker.com/develop/develop-images/dockerfile_best-practices/
3. FastAPI 部署文档：https://fastapi.tiangolo.com/deployment/
4. GitHub Actions 文档：https://docs.github.com/en/actions