api_db_integration

API + БД: сквозные production-сценарии

Архитектура: API контейнер + PostgreSQL контейнер. Сквозные тесты: POST → GET → проверка консистентности. Изоляция: сброс состояния БД между API-тестами. Негативные сценарии: недоступная БД

Реальные production-тесты проверяют полный стек: API получает запрос → пишет в БД → читает из БД → возвращает ответ.

Цель: Научиться тестировать сквозные сценарии с несколькими контейнерами и изолировать тесты друг от друга.

Архитектура: два контейнера

┌──────────────────┐         ┌──────────────────┐
│  FastAPI         │         │  PostgreSQL      │
│  Container       │ ──────> │  Container       │
│  Port: 8000      │         │  Port: 5432      │
└──────────────────┘         └──────────────────┘
        ↑
        │ HTTP requests
        │
┌──────────────────┐
│  Pytest Tests    │
└──────────────────┘

Обновленный FastAPI с PostgreSQL

# api/main.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import psycopg2
import os

app = FastAPI()

# Подключение к PostgreSQL
def get_db_connection():
    return psycopg2.connect(
        host=os.getenv("DB_HOST", "localhost"),
        port=os.getenv("DB_PORT", "5432"),
        user=os.getenv("DB_USER", "testuser"),
        password=os.getenv("DB_PASSWORD", "testpass"),
        dbname=os.getenv("DB_NAME", "testdb")
    )

class TodoCreate(BaseModel):
    title: str
    description: str

class TodoResponse(BaseModel):
    id: int
    title: str
    description: str

@app.post("/todos", response_model=TodoResponse)
def create_todo(todo: TodoCreate):
    """Создать todo в БД"""
    conn = get_db_connection()
    cur = conn.cursor()

    cur.execute(
        "INSERT INTO todos (title, description) VALUES (%s, %s) RETURNING id, title, description",
        (todo.title, todo.description)
    )
    result = cur.fetchone()
    conn.commit()
    cur.close()
    conn.close()

    return {"id": result[0], "title": result[1], "description": result[2]}

@app.get("/todos", response_model=list[TodoResponse])
def list_todos():
    """Список todos из БД"""
    conn = get_db_connection()
    cur = conn.cursor()

    cur.execute("SELECT id, title, description FROM todos ORDER BY id")
    results = cur.fetchall()

    cur.close()
    conn.close()

    return [{"id": row[0], "title": row[1], "description": row[2]} for row in results]

@app.get("/todos/{todo_id}", response_model=TodoResponse)
def get_todo(todo_id: int):
    """Получить один todo"""
    conn = get_db_connection()
    cur = conn.cursor()

    cur.execute("SELECT id, title, description FROM todos WHERE id = %s", (todo_id,))
    result = cur.fetchone()

    cur.close()
    conn.close()

    if not result:
        raise HTTPException(status_code=404, detail="Todo not found")

    return {"id": result[0], "title": result[1], "description": result[2]}

docker-compose для тестов

# tests/docker-compose.test.yml

version: "3.8"

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: testuser
      POSTGRES_PASSWORD: testpass
      POSTGRES_DB: testdb
    ports:
      - "5432:5432"

  api:
    build: ../api
    depends_on:
      - postgres
    environment:
      DB_HOST: postgres
      DB_PORT: 5432
      DB_USER: testuser
      DB_PASSWORD: testpass
      DB_NAME: testdb
    ports:
      - "8000:8000"

Фикстура для сквозных тестов

# tests/conftest.py

import pytest
import subprocess
import time
import requests
import psycopg2

@pytest.fixture(scope="session")
def docker_services():
    """Запустить docker-compose со всеми сервисами"""
    # Запустить контейнеры
    subprocess.run(["docker-compose", "-f", "tests/docker-compose.test.yml", "up", "-d"], check=True)

    # Подождать пока PostgreSQL готов
    time.sleep(5)

    # Применить миграции
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        user="testuser",
        password="testpass",
        dbname="testdb"
    )
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS todos (
            id SERIAL PRIMARY KEY,
            title VARCHAR(255) NOT NULL,
            description TEXT
        )
    """)
    conn.commit()
    cur.close()
    conn.close()

    # Подождать пока API готов
    for _ in range(30):
        try:
            response = requests.get("http://localhost:8000/todos")
            if response.status_code == 200:
                break
        except:
            time.sleep(1)

    yield

    # Остановить после тестов
    subprocess.run(["docker-compose", "-f", "tests/docker-compose.test.yml", "down", "-v"], check=True)

@pytest.fixture
def clean_db():
    """Очистить БД перед каждым тестом"""
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        user="testuser",
        password="testpass",
        dbname="testdb"
    )
    cur = conn.cursor()
    cur.execute("TRUNCATE TABLE todos RESTART IDENTITY CASCADE")
    conn.commit()
    cur.close()
    conn.close()

    yield

@pytest.fixture
def api_url(docker_services):
    """URL для API"""
    return "http://localhost:8000"

Сквозной тест: POST → GET

# tests/test_todo_e2e.py

import requests

def test_create_and_retrieve_todo(api_url, clean_db):
    """Сквозной тест: создать и получить todo"""
    # Создать todo
    response = requests.post(
        f"{api_url}/todos",
        json={"title": "Test Todo", "description": "Test Description"}
    )

    assert response.status_code == 200
    created_todo = response.json()
    assert created_todo["id"] == 1
    assert created_todo["title"] == "Test Todo"

    # Получить todo по ID
    response = requests.get(f"{api_url}/todos/{created_todo['id']}")

    assert response.status_code == 200
    retrieved_todo = response.json()
    assert retrieved_todo["id"] == created_todo["id"]
    assert retrieved_todo["title"] == "Test Todo"
    assert retrieved_todo["description"] == "Test Description"

def test_list_multiple_todos(api_url, clean_db):
    """Тест списка todos"""
    # Создать 3 todos
    for i in range(1, 4):
        requests.post(
            f"{api_url}/todos",
            json={"title": f"Todo {i}", "description": f"Description {i}"}
        )

    # Получить список
    response = requests.get(f"{api_url}/todos")

    assert response.status_code == 200
    todos = response.json()
    assert len(todos) == 3
    assert todos[0]["title"] == "Todo 1"
    assert todos[1]["title"] == "Todo 2"
    assert todos[2]["title"] == "Todo 3"

Негативный сценарий: todo не существует

def test_get_nonexistent_todo(api_url, clean_db):
    """Тест получения несуществующего todo"""
    response = requests.get(f"{api_url}/todos/999")

    assert response.status_code == 404
    assert response.json()["detail"] == "Todo not found"

Проверка консистентности БД

def test_db_consistency_after_create(api_url, clean_db):
    """Тест что данные действительно записались в БД"""
    # Создать todo через API
    response = requests.post(
        f"{api_url}/todos",
        json={"title": "Consistency Test", "description": "Test"}
    )
    todo_id = response.json()["id"]

    # Проверить напрямую в БД
    import psycopg2
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        user="testuser",
        password="testpass",
        dbname="testdb"
    )
    cur = conn.cursor()
    cur.execute("SELECT title, description FROM todos WHERE id = %s", (todo_id,))
    result = cur.fetchone()
    cur.close()
    conn.close()

    assert result is not None
    assert result[0] == "Consistency Test"
    assert result[1] == "Test"

Проверьте свои знания

Вопросы ещё не добавлены

Вопросы для этой подтемы ещё не добавлены.

Далее: Advanced fixtures: patterns и best practices

API + БД: сквозные production-сценарии

┌──────────────────┐ ┌──────────────────┐ │ FastAPI │ │ PostgreSQL │ │ Container │ ──────> │ Container │ │ Port: 8000 │ │ Port: 5432 │ └──────────────────┘ └──────────────────┘ ↑ │ HTTP requests │ ┌──────────────────┐ │ Pytest Tests │ └──────────────────┘