postgresql_advanced

PostgreSQL advanced: сложные сценарии

Connection pooling в тестах: когда нужен, как настроить. Тестирование constraints и FK. Изоляция тестов: TRUNCATE vs DELETE. Проблемы sequence и их решение

В этом уроке углубимся в сложные сценарии: связанные таблицы, foreign keys, connection pooling. Научимся тестировать реальные бизнес-кейсы.

Цель урока: Тестировать сложные структуры данных и управлять тестовыми данными эффективно.

Множественные таблицы с Foreign Keys

Реальные приложения используют связанные таблицы. Например, блог:

# conftest.py - добавьте к существующему db_setup
@pytest.fixture(scope="session")
def db_setup():
    """Создаём связанные таблицы"""
    conn = psycopg2.connect(...)
    cur = conn.cursor()

    # Удаляем существующие таблицы (CASCADE удаляет зависимые)
    cur.execute("DROP TABLE IF EXISTS comments CASCADE")
    cur.execute("DROP TABLE IF EXISTS posts CASCADE")
    cur.execute("DROP TABLE IF EXISTS users CASCADE")

    # Создаём таблицы
    cur.execute("""
        CREATE TABLE users (
            id SERIAL PRIMARY KEY,
            email VARCHAR(255) UNIQUE NOT NULL,
            name VARCHAR(255)
        )
    """)

    cur.execute("""
        CREATE TABLE posts (
            id SERIAL PRIMARY KEY,
            user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            title VARCHAR(255) NOT NULL,
            content TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cur.execute("""
        CREATE TABLE comments (
            id SERIAL PRIMARY KEY,
            post_id INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
            user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            text TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    conn.commit()
    cur.close()
    conn.close()

    yield

    # Cleanup
    conn = psycopg2.connect(...)
    cur = conn.cursor()
    cur.execute("DROP TABLE IF EXISTS comments CASCADE")
    cur.execute("DROP TABLE IF EXISTS posts CASCADE")
    cur.execute("DROP TABLE IF EXISTS users CASCADE")
    conn.commit()
    cur.close()
    conn.close()

Важно:

REFERENCES users(id) — foreign key к таблице users
ON DELETE CASCADE — при удалении user удаляются его posts и comments
Порядок создания важен: сначала родители (users), потом дети (posts, comments)

Тестирование Foreign Key constraints

# test_foreign_keys.py
import pytest
import psycopg2

def test_cannot_create_post_without_user(db_connection):
    """Нельзя создать post без существующего user_id"""
    cur = db_connection.cursor()

    # Пытаемся создать post с несуществующим user_id
    with pytest.raises(psycopg2.errors.ForeignKeyViolation):
        cur.execute(
            "INSERT INTO posts (user_id, title) VALUES (%s, %s)",
            (9999, "Test Post")  # user_id=9999 не существует
        )

    cur.close()

def test_cascade_delete(db_connection):
    """При удалении user удаляются его posts"""
    cur = db_connection.cursor()

    # Создаём user
    cur.execute("INSERT INTO users (email, name) VALUES (%s, %s) RETURNING id", ("alice@test.com", "Alice"))
    user_id = cur.fetchone()[0]

    # Создаём 2 posts
    cur.execute("INSERT INTO posts (user_id, title) VALUES (%s, %s)", (user_id, "Post 1"))
    cur.execute("INSERT INTO posts (user_id, title) VALUES (%s, %s)", (user_id, "Post 2"))

    # Проверяем что posts созданы
    cur.execute("SELECT COUNT(*) FROM posts WHERE user_id=%s", (user_id,))
    assert cur.fetchone()[0] == 2

    # Удаляем user
    cur.execute("DELETE FROM users WHERE id=%s", (user_id,))

    # Проверяем что posts тоже удалились (CASCADE)
    cur.execute("SELECT COUNT(*) FROM posts WHERE user_id=%s", (user_id,))
    assert cur.fetchone()[0] == 0

    cur.close()

Helper-функции для связанных данных

Создадим helper'ы для удобного создания тестовых данных:

# conftest.py - добавьте helpers
def create_user(conn, email, name="Test User"):
    """Создаёт user"""
    cur = conn.cursor()
    cur.execute("INSERT INTO users (email, name) VALUES (%s, %s) RETURNING id", (email, name))
    user_id = cur.fetchone()[0]
    cur.close()
    return user_id

def create_post(conn, user_id, title, content="Test content"):
    """Создаёт post"""
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO posts (user_id, title, content) VALUES (%s, %s, %s) RETURNING id",
        (user_id, title, content)
    )
    post_id = cur.fetchone()[0]
    cur.close()
    return post_id

def create_comment(conn, post_id, user_id, text):
    """Создаёт comment"""
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO comments (post_id, user_id, text) VALUES (%s, %s, %s) RETURNING id",
        (post_id, user_id, text)
    )
    comment_id = cur.fetchone()[0]
    cur.close()
    return comment_id

Использование:

def test_blog_scenario(db_connection):
    """Полный сценарий: user создаёт post, другой комментирует"""
    # Создаём двух users
    alice_id = create_user(db_connection, "alice@test.com", "Alice")
    bob_id = create_user(db_connection, "bob@test.com", "Bob")

    # Alice создаёт post
    post_id = create_post(db_connection, alice_id, "My first post")

    # Bob комментирует
    comment_id = create_comment(db_connection, post_id, bob_id, "Great post!")

    # Проверяем связи
    cur = db_connection.cursor()
    cur.execute("""
        SELECT u.name, p.title, c.text
        FROM comments c
        JOIN posts p ON c.post_id = p.id
        JOIN users u ON c.user_id = u.id
        WHERE c.id = %s
    """, (comment_id,))

    result = cur.fetchone()
    assert result == ("Bob", "My first post", "Great post!")
    cur.close()

Стратегии тестовых данных

Стратегия 1: Минимальные данные

Идея: Создавать минимум данных для теста.

def test_user_can_create_post(db_connection):
    """Тестируем только создание post"""
    user_id = create_user(db_connection, "test@example.com")
    post_id = create_post(db_connection, user_id, "Test")

    cur = db_connection.cursor()
    cur.execute("SELECT title FROM posts WHERE id=%s", (post_id,))
    assert cur.fetchone()[0] == "Test"
    cur.close()

Плюсы:

✅ Тест быстрый
✅ Легко понять что тестируем

Стратегия 2: Сложные сценарии

Идея: Создавать реалистичные данные для edge cases.

def test_user_with_many_posts(db_connection):
    """User с 100 posts"""
    user_id = create_user(db_connection, "prolific@test.com")

    # Создаём 100 posts
    for i in range(100):
        create_post(db_connection, user_id, f"Post {i}")

    cur = db_connection.cursor()
    cur.execute("SELECT COUNT(*) FROM posts WHERE user_id=%s", (user_id,))
    assert cur.fetchone()[0] == 100
    cur.close()

Стратегия 3: Фикстуры с предзаполненными данными

Идея: Создать фикстуру с базовым набором данных.

# conftest.py
@pytest.fixture
def blog_with_data(db_connection):
    """Фикстура с готовым блогом"""
    # Создаём 3 users
    users = [
        create_user(db_connection, f"user{i}@test.com", f"User{i}")
        for i in range(1, 4)
    ]

    # Каждый user создаёт 2 posts
    posts = []
    for user_id in users:
        for j in range(2):
            post_id = create_post(db_connection, user_id, f"Post by user {user_id}-{j}")
            posts.append(post_id)

    return {"users": users, "posts": posts}

# test_with_preloaded_data.py
def test_total_posts(blog_with_data, db_connection):
    """Проверяем количество posts"""
    cur = db_connection.cursor()
    cur.execute("SELECT COUNT(*) FROM posts")
    assert cur.fetchone()[0] == 6  # 3 users * 2 posts
    cur.close()

Connection Pooling (опционально для production)

В production используйте connection pool для эффективности:

# conftest.py
from psycopg2 import pool

@pytest.fixture(scope="session")
def db_pool():
    """Connection pool для тестов"""
    connection_pool = pool.SimpleConnectionPool(
        minconn=1,
        maxconn=10,
        host="localhost",
        port=5433,
        user="testuser",
        password="testpass",
        database="testdb"
    )
    yield connection_pool
    connection_pool.closeall()

@pytest.fixture
def db_connection(db_pool):
    """Берём connection из pool"""
    conn = db_pool.getconn()
    yield conn
    conn.rollback()
    db_pool.putconn(conn)  # Возвращаем в pool

Преимущества:

✅ Быстрее (переиспользуем connections)
✅ Меньше нагрузка на PostgreSQL

Для тестов: Обычный psycopg2.connect() достаточен, pool нужен при >100 тестах.

Тестирование сложных запросов

def test_user_post_count(db_connection):
    """Проверяем подсчёт posts для каждого user"""
    # Создаём users с разным количеством posts
    alice_id = create_user(db_connection, "alice@test.com")
    bob_id = create_user(db_connection, "bob@test.com")

    # Alice: 3 posts, Bob: 1 post
    for i in range(3):
        create_post(db_connection, alice_id, f"Alice post {i}")
    create_post(db_connection, bob_id, "Bob post")

    # Запрос с GROUP BY
    cur = db_connection.cursor()
    cur.execute("""
        SELECT u.name, COUNT(p.id) as post_count
        FROM users u
        LEFT JOIN posts p ON u.id = p.user_id
        GROUP BY u.id, u.name
        ORDER BY u.name
    """)
    results = cur.fetchall()

    assert results == [("Alice", 3), ("Bob", 1)]
    cur.close()

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

def test_unique_post_title_per_user(db_connection):
    """Проверяем бизнес-правило: user не может создать 2 posts с одинаковым title"""
    # Сначала добавьте UNIQUE constraint в db_setup:
    # CREATE UNIQUE INDEX unique_user_post_title ON posts(user_id, title)

    user_id = create_user(db_connection, "user@test.com")

    # Создаём первый post
    create_post(db_connection, user_id, "Unique Title")

    # Пытаемся создать дубликат
    with pytest.raises(psycopg2.errors.UniqueViolation):
        create_post(db_connection, user_id, "Unique Title")

Что вы узнали

✅ Foreign Keys — связи между таблицами, тестирование constraints ✅ CASCADE операции — автоматическое удаление связанных данных ✅ Helper-функции — создание связанных тестовых данных ✅ Стратегии данных — минимальные, сложные, предзаполненные ✅ Connection Pool — переиспользование connections (опционально) ✅ Сложные запросы — JOIN, GROUP BY, тестирование аггрегаций

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

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

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

Далее: REST API: тестирование с Testcontainers