Синхронное тестирование БД: фундамент для начинающих

Этот материал был частью продвинутого курса по pytest, но был вынесен отдельно. Если вы ещё не знаете async/await, начните с этого гайда перед переходом к async-тестированию.

Для кого этот материал?

✅ Вы знаете pytest (fixtures, mocks, markers)

✅ Вы работаете с PostgreSQL

❌ Вы НЕ знаете async/await

❌ Вы не готовы к асинхронному коду

Синхронные фикстуры для PostgreSQL

Проблема: test pollution

Наивный подход:

import psycopg2
 
def test_create_user():
    conn = psycopg2.connect(
        "postgresql://postgres:testpass@localhost:5432/test_db"
    )
    cur = conn.cursor()
    cur.execute("INSERT INTO users (email) VALUES ('user@test.com')")
    conn.commit()  # ❌ Данные остаются в БД!
    conn.close()
 
def test_user_count():
    conn = psycopg2.connect(...)
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM users")
    count = cur.fetchone()[0]
    assert count == 0  # ❌ FAIL! Есть user@test.com из предыдущего теста

Проблема: Тесты влияют друг на друга.

Решение: транзакции с rollback

# conftest.py
import pytest
import psycopg2
 
@pytest.fixture(scope="function")
def db_connection():
    """Connection с автоматическим rollback"""
    conn = psycopg2.connect(
        "postgresql://postgres:testpass@localhost:5432/test_db"
    )
    # Транзакция начинается автоматически
    yield conn
    # После теста: откатываем ВСЁ
    conn.rollback()
    conn.close()

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

def test_create_user(db_connection):
    cur = db_connection.cursor()
    cur.execute("INSERT INTO users (email) VALUES ('user@test.com')")
    # НЕТ commit()! Изменения только в транзакции
 
    cur.execute("SELECT email FROM users WHERE email='user@test.com'")
    result = cur.fetchone()
    assert result[0] == "user@test.com"
    # После теста: rollback, данные исчезнут
 
def test_user_count(db_connection):
    cur = db_connection.cursor()
    cur.execute("SELECT COUNT(*) FROM users")
    count = cur.fetchone()[0]
    assert count == 0  # ✅ PASS! Чистая БД

Race Conditions: threading

Если вы хотите понять race conditions без async, вот простой пример:

import threading
 
counter = 0
 
def increment():
    global counter
    current = counter
    import time
    time.sleep(0.0001)  # Задержка
    counter = current + 1
 
def test_race_condition():
    """Воспроизводим гонку"""
    global counter
    counter = 0
 
    threads = [threading.Thread(target=increment) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
 
    assert counter == 10  # ❌ FAIL! counter = 3-7

Исправление: threading.Lock

import threading
 
counter = 0
lock = threading.Lock()
 
def increment_safe():
    global counter
    with lock:
        current = counter
        import time
        time.sleep(0.0001)
        counter = current + 1
 
def test_no_race():
    global counter
    counter = 0
 
    threads = [threading.Thread(target=increment_safe) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
 
    assert counter == 10  # ✅ PASS!