python-integration（Python 集成）

Python Integration Testing — Core Patterns

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: python-integration-testing for comprehensive documentation on patterns, SQLAlchemy fixtures, Alembic, and CI/CD.

Test Pyramid (70/20/10 Rule)

         ╔══════════╗
         ║   E2E    ║  10% — Playwright, Selenium, full stack
         ╠══════════╣
         ║Integration║  20% — Real DB, real services, API flows
         ╠══════════╣
         ║   Unit   ║  70% — Fast, isolated, no I/O
         ╚══════════╝

Integration tests touch at least 2 real components:

• Application code + real database
• API endpoint + real dependency
• Message producer + real broker

conftest.py Architecture

tests/
├── conftest.py              # Session: containers, engines
├── unit/
│   ├── conftest.py          # Unit-only fixtures
│   └── test_services.py
├── integration/
│   ├── conftest.py          # DB sessions, HTTP clients
│   ├── test_api.py
│   └── test_repositories.py
└── e2e/
    ├── conftest.py
    └── test_flows.py

# tests/conftest.py — session-level infrastructure
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg

@pytest.fixture(scope="session")
def redis_container():
    with RedisContainer("redis:7-alpine") as redis:
        yield redis

# tests/integration/conftest.py — per-test isolation
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="module")
def engine(postgres_container):
    url = postgres_container.get_connection_url()
    eng = create_engine(url)
    Base.metadata.create_all(eng)
    yield eng
    Base.metadata.drop_all(eng)

@pytest.fixture
def db_session(engine):
    """Savepoint rollback — fast and safe."""
    conn = engine.connect()
    trans = conn.begin()
    session = sessionmaker(bind=conn)()
    nested = conn.begin_nested()
    yield session
    session.close()
    nested.rollback()
    trans.rollback()
    conn.close()

pytest Markers

# pyproject.toml
[tool.pytest.ini_options]
markers = [
    "integration: integration tests (require Docker)",
    "slow: slow tests (> 5 seconds)",
    "e2e: end-to-end tests",
]

# Mark tests
@pytest.mark.integration
def test_user_persisted(db_session):
    ...

# Skip if no Docker
import shutil
requires_docker = pytest.mark.skipif(
    not shutil.which("docker"),
    reason="Docker not available"
)

@pytest.mark.integration
@requires_docker
def test_with_container():
    ...

# Run only integration tests
pytest -m integration

# Exclude slow tests
pytest -m "integration and not slow"

# Run everything except e2e
pytest -m "not e2e"

GitHub Actions — Testcontainers vs Service Containers

Testcontainers (recommended): containers managed by the test code itself.

# .github/workflows/integration-tests.yml
name: Integration Tests
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -e ".[test]"
      - run: pytest -m integration -v
        env:
          DOCKER_HOST: unix:///var/run/docker.sock

Service containers (simpler for single-service tests):

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_DB: testdb
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports: ["5432:5432"]
    steps:
      - run: pytest -m integration
        env:
          DATABASE_URL: postgresql://test:test@localhost:5432/testdb

Parallel Execution — pytest-xdist

pip install pytest-xdist

pytest -n 4        # 4 workers
pytest -n auto     # one per CPU core

Database isolation strategies with xdist:

# 1. Database per worker (most isolated)
@pytest.fixture(scope="session")
def db_url(worker_id, tmp_path_factory):
    if worker_id == "master":
        return create_db("test_db")
    return create_db(f"test_db_{worker_id}")

# 2. PostgreSQL schema per worker
@pytest.fixture(scope="session")
def db_session(worker_id, engine):
    schema = f"test_{worker_id}"
    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema}"))
    yield ...

# 3. Transaction rollback (works without worker isolation)
@pytest.fixture
def db_session(engine):
    conn = engine.connect()
    conn.begin()
    conn.begin_nested()  # SAVEPOINT
    yield sessionmaker(bind=conn)()
    conn.rollback()
    conn.close()

Test Isolation Comparison

Anti-Patterns

Reference Documentation

• Integration Patterns
• SQLAlchemy Fixtures
• Alembic Testing

Official docs:

• pytest: https://docs.pytest.org/
• pytest-xdist: https://pytest-xdist.readthedocs.io/