集成测试规范

集成测试规范

接口、模块集成测试

---

测试范围

规则

• [Requirement] 集成测试验证多个模块协作的正确性，覆盖模块间接口、数据流转、事务边界；不得退化为单元测试或端到端测试。
• [Requirement] 集成测试必须使用独立的测试环境，禁止连接生产数据库或生产外部服务。
• [Standard] 集成测试的范围边界：最多覆盖”应用 + 数据库 + 消息队列 + 缓存”，外部第三方服务（支付、短信、第三方 API）仍需 Mock。
• [Standard] 集成测试与单元测试物理隔离：单独目录 tests/integration/，单独 marker @pytest.mark.integration，CI 中独立 Stage 执行。
• [Default] 集成测试聚焦”正常流程 + 关键异常流程”，不追求穷举边界；边界值由单元测试覆盖。
• [Suggestion] 按业务场景（use case）组织集成测试，而非按类/方法；一个测试对应一个完整业务动作。
• [Caution] 集成测试不得依赖测试执行顺序；每个测试必须能独立运行，自行准备和清理数据。

代码示例

# 目录结构
tests/
├── unit/                  # 单元测试
│   └── services/
└── integration/           # 集成测试
    ├── conftest.py        # 测试环境 fixture（DB、Redis、MQ）
    ├── test_user_api.py
    └── test_order_flow.py

# tests/integration/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
 
TEST_DB_URL = "postgresql://test:test@localhost:5432/testdb"
 
 
@pytest.fixture(scope="session")
def engine():
    engine = create_engine(TEST_DB_URL)
    yield engine
    engine.dispose()
 
 
@pytest.fixture
def db_session(engine):
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.rollback()
    session.close()

反模式

# 反模式：集成测试连接生产环境
TEST_DB_URL = "postgresql://user:pass@prod-db:5432/proddb"
 
# 反模式：集成测试依赖执行顺序
def test_create_user():
    client.post("/users", json={"name": "Alice"})
 
def test_get_user():
    resp = client.get("/users/1")  # 假设 test_create_user 先执行
    assert resp.json()["name"] == "Alice"
 
# 反模式：集成测试目录与单元测试混在一起
tests/
├── test_user_service.py      # 单元测试
└── test_user_api.py          # 集成测试，无 marker，无法分离执行

---

接口测试

规则

• [Requirement] 所有对外 HTTP API 必须有集成测试覆盖，验证请求体校验、响应状态码、响应体结构、错误码。
• [Requirement] 接口测试必须覆盖：正常请求、参数缺失、参数类型错误、权限不足、资源不存在、服务端异常。
• [Standard] 使用真实的 HTTP 客户端（如 httpx.AsyncClient、requests）发起请求，不直接调用路由函数，确保中间件、鉴权、序列化链路完整。
• [Standard] 响应体校验使用契约校验（如 pydantic 模型），而非逐字段断言，避免接口变更时测试维护成本过高。
• [Default] 接口测试断言优先级：HTTP 状态码 → 业务错误码 → 响应体关键字段。
• [Suggestion] 对分页接口，测试空列表、单页满、跨页三种场景。

代码示例

# tests/integration/test_user_api.py
import pytest
from httpx import AsyncClient
 
from myapp.main import app
from myapp.schemas import UserOut
 
 
@pytest.mark.asyncio
async def test_create_user_success(client: AsyncClient) -> None:
    payload = {"name": "Alice", "email": "alice@example.com"}
 
    resp = await client.post("/api/v1/users", json=payload)
 
    assert resp.status_code == 201
    body = resp.json()
    assert body["code"] == 0
    # 契约校验
    UserOut.model_validate(body["data"])
 
 
@pytest.mark.asyncio
async def test_create_user_missing_email_returns_400(client: AsyncClient) -> None:
    resp = await client.post("/api/v1/users", json={"name": "Alice"})
 
    assert resp.status_code == 400
    assert resp.json()["code"] == 40001  # 参数校验错误码
 
 
@pytest.mark.asyncio
async def test_get_user_not_found_returns_404(client: AsyncClient) -> None:
    resp = await client.get("/api/v1/users/999999")
 
    assert resp.status_code == 404
    assert resp.json()["code"] == 40401

反模式

# 反模式：直接调用路由函数，跳过中间件
def test_create_user():
    router = get_user_router()
    result = router.create_user({"name": "Alice"})  # 绕过鉴权、日志中间件
 
# 反模式：只测正常流程，不测异常
def test_create_user():
    resp = client.post("/users", json={"name": "Alice"})
    assert resp.status_code == 201
    # 没有测参数错误、权限不足等场景
 
# 反模式：断言响应体全部字段，维护成本高
def test_get_user():
    resp = client.get("/users/1")
    body = resp.json()
    assert body["id"] == 1
    assert body["name"] == "Alice"
    assert body["email"] == "alice@example.com"
    assert body["created_at"] == "2024-01-01T00:00:00Z"  # 时间字段每次都变

---

模块集成

规则

• [Requirement] 涉及多模块协作的核心业务流程（如下单→扣库存→生成订单→发消息）必须有集成测试，验证模块间接口契约。
• [Standard] 模块集成测试验证：数据在模块间流转的正确性、事务一致性（提交/回滚）、事件发布与消费。
• [Standard] 跨服务集成测试只测本服务边界，下游服务通过 Mock Server（如 WireMock、responses）模拟，不依赖真实下游。
• [Default] 消息队列集成测试使用真实的测试 MQ（如本地 Redis/RabbitMQ），验证消息生产、消费、重试、死信逻辑。
• [Suggestion] 对长链路流程，用 BDD 风格描述场景（Given-When-Then），提升可读性。

代码示例

# tests/integration/test_order_flow.py
import pytest
from myapp.services import OrderService
from myapp.repositories import OrderRepository, InventoryRepository
 
 
@pytest.fixture
def order_repo(db_session) -> OrderRepository:
    return OrderRepository(db_session)
 
 
@pytest.fixture
def inventory_repo(db_session) -> InventoryRepository:
    return InventoryRepository(db_session)
 
 
def test_create_order_deducts_inventory_and_publishes_event(
    order_repo, inventory_repo, message_bus, db_session
) -> None:
    # Given: 库存充足
    inventory_repo.add(product_id=1, stock=10)
    db_session.commit()
 
    # When: 创建订单
    service = OrderService(order_repo, inventory_repo, message_bus)
    order = service.create_order(user_id=1, product_id=1, quantity=3)
 
    # Then: 库存扣减
    assert inventory_repo.get_stock(1) == 7
    # And: 订单已持久化
    assert order_repo.get_by_id(order.id) is not None
    # And: 事件已发布
    messages = message_bus.get_published("order.created")
    assert len(messages) == 1
    assert messages[0].order_id == order.id

反模式

# 反模式：跨服务集成测试依赖真实下游
def test_create_order():
    service = OrderService(inventory_client=RealInventoryClient())  # 调真实库存服务
    order = service.create_order(product_id=1, quantity=1)
    # 网络抖动、下游不稳定都会导致测试失败
 
# 反模式：只验证最终状态，不验证中间链路
def test_order_flow():
    create_order(product_id=1, quantity=3)
    assert get_order(1).status == "created"
    # 没有验证库存是否扣减、消息是否发送

---

测试数据

规则

• [Requirement] 集成测试数据必须由测试自行准备（Fixture/Factory），测试结束后必须清理，不得残留到数据库中。
• [Requirement] 禁止依赖测试环境中已有的”脏数据”；每个测试的前置数据在测试内显式创建。
• [Standard] 数据清理策略：优先使用事务回滚（每个测试开事务，结束 rollback）；无法回滚的（如发消息）用显式 delete。
• [Standard] 测试数据使用确定性值（固定 ID、固定时间），避免使用自增 ID 或当前时间导致断言困难。
• [Default] 多个测试共用的基础数据提取为 session-scoped fixture，单测试专用数据用 function-scoped fixture。
• [Suggestion] 使用 Factory Boy 或类似库构造测试数据，避免在每个测试中手写大量 INSERT。

代码示例

# tests/integration/conftest.py
import pytest
from myapp.models import User
 
 
@pytest.fixture(autouse=True)
def clean_db(db_session):
    """每个测试后清理所有表"""
    yield
    db_session.execute("TRUNCATE TABLE users, orders, inventories RESTART IDENTITY CASCADE")
    db_session.commit()
 
 
@pytest.fixture
def test_user(db_session) -> User:
    user = User(id=1001, name="Alice", email="alice@example.com")
    db_session.add(user)
    db_session.commit()
    return user

# 使用 Factory 构造数据
# tests/factories.py
import factory
from myapp.models import User, Order
 
 
class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = db_session
 
    id = factory.Sequence(lambda n: 1000 + n)
    name = factory.Faker("name")
    email = factory.Faker("email")
 
 
class OrderFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = Order
        sqlalchemy_session = db_session
 
    id = factory.Sequence(lambda n: 2000 + n)
    user = factory.SubFactory(UserFactory)
    amount = factory.Faker("pydecimal", left_digits=5, right_digits=2, positive=True)

反模式

# 反模式：依赖环境已有数据
def test_get_user():
    resp = client.get("/users/1")  # 假设 id=1 的用户已存在
    assert resp.status_code == 200
    # 换一个干净环境就挂了
 
# 反模式：测试后不清理数据
def test_create_user():
    client.post("/users", json={"name": "Alice"})
    assert resp.status_code == 201
    # 数据残留，下次运行可能主键冲突或数据污染
 
# 反模式：使用自增 ID 做断言
def test_create_user():
    resp = client.post("/users", json={"name": "Alice"})
    assert resp.json()["id"] == 1  # 自增 ID 不稳定

---

环境

规则

• [Requirement] 集成测试环境必须与生产环境技术栈一致（数据库版本、中间件版本），禁止用 H2 替代 Postgres、用内存缓存替代 Redis。
• [Requirement] 集成测试环境必须完全隔离：独立数据库、独立 Schema、独立端口，不得与开发/测试环境共用。
• [Standard] 使用 Docker Compose 一键启动测试依赖（数据库、Redis、MQ），保证团队成员和 CI 环境一致。
• [Standard] 集成测试超时设置：单测试不超过 30 秒，整组不超过 10 分钟；超时的测试标记为失败。
• [Default] CI 中集成测试在单元测试通过后执行；失败时阻断合并。
• [Suggestion] 本地开发可通过 make test-integration 一键拉起依赖并执行测试。
• [Caution] 禁止在集成测试中连接任何生产或预发资源，包括共享的测试数据库。

代码示例

# tests/integration/docker-compose.yml
version: "3.9"
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: test
      POSTGRES_PASSWORD: test
      POSTGRES_DB: testdb
    ports:
      - "5433:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U test"]
      interval: 5s
      timeout: 5s
      retries: 10
 
  redis:
    image: redis:7-alpine
    ports:
      - "6380:6379"
 
  rabbitmq:
    image: rabbitmq:3.12-management-alpine
    environment:
      RABBITMQ_DEFAULT_USER: test
      RABBITMQ_DEFAULT_PASS: test
    ports:
      - "5673:5672"

# Makefile
.PHONY: test-integration
test-integration:
	docker compose -f tests/integration/docker-compose.yml up -d
	@pytest tests/integration/ -m integration --maxfail=1
	@docker compose -f tests/integration/docker-compose.yml down

# pytest.ini 超时配置
[pytest]
markers =
    integration: marks tests as integration tests
asyncio_mode = auto
timeout = 30

反模式

# 反模式：用 H2 替代 Postgres，SQL 方言不一致导致测试通过生产失败
services:
  db:
    image: h2database/h2:latest  # 行为与 Postgres 不同
 
# 反模式：连接共享测试库
TEST_DB_URL=postgresql://shared:shared@team-db:5432/shared_test
# 多人同时跑测试会互相干扰，数据被覆盖