Фундамент: pytest-asyncio и окружение

# test_async_knowledge.py
import asyncio
 
async def check_knowledge():
    """Вы ДОЛЖНЫ понимать этот код"""
    tasks = [asyncio.sleep(0.1) for _ in range(5)]
    await asyncio.gather(*tasks)
 
    lock = asyncio.Lock()
    async with lock:
        print("Critical section")
 
check_knowledge()  # Если не понимаете — курс НЕ для вас!

# Установите Docker Desktop
brew install --cask docker
# Запустите Docker Desktop из Applications

curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh
sudo usermod -aG docker $USER
# Перелогиньтесь!

docker --version
# Docker version 24.0.0 или выше

docker run -d \
  --name pytest-postgres \
  -e POSTGRES_PASSWORD=testpass \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_DB=pytest_test \
  -p 5432:5432 \
  postgres:15-alpine
 
# Проверка
docker exec pytest-postgres pg_isready
# /var/run/postgresql:5432 - accepting connections

docker exec -i pytest-postgres psql -U postgres -d pytest_test <<EOF
CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);
EOF

docker run -d \
  --name pytest-redis \
  -p 6379:6379 \
  redis:7-alpine
 
# Проверка
docker exec pytest-redis redis-cli ping
# PONG

# Создайте виртуальное окружение
python3 -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
 
# Установите зависимости
pip install pytest pytest-asyncio asyncpg redis[asyncio]

# test_deps.py
import pytest
import asyncpg
import redis.asyncio as aioredis
 
print("✅ All deps installed")

python test_deps.py
# ✅ All deps installed

# pytest.ini
[pytest]
asyncio_mode = auto

# tests/test_first_async.py
import pytest
 
@pytest.mark.asyncio
async def test_simple_async():
    """Простейший async тест"""
    import asyncio
    await asyncio.sleep(0.001)
    assert True

pytest tests/test_first_async.py -v

test_simple_async PASSED

# conftest.py
import pytest
 
@pytest.fixture
async def async_value():
    """Async фикстура возвращает значение"""
    import asyncio
    await asyncio.sleep(0.001)  # Имитация async операции
    return 42
 
@pytest.mark.asyncio
async def test_async_fixture(async_value):
    """Тест использует async фикстуру"""
    assert async_value == 42

# conftest.py
import pytest
 
@pytest.fixture
async def async_resource():
    """Async фикстура с teardown"""
    # Setup
    print("Opening async resource")
    resource = {"status": "open"}
 
    yield resource  # Передаём в тест
 
    # Teardown
    print("Closing async resource")
    resource["status"] = "closed"

@pytest.mark.asyncio
async def test_resource(async_resource):
    assert async_resource["status"] == "open"
    # После теста выполнится teardown

# conftest.py
import pytest
import asyncpg
 
@pytest.fixture
async def db_connection():
    """Async connection к PostgreSQL"""
    conn = await asyncpg.connect(
        host="localhost",
        port=5432,
        user="postgres",
        password="testpass",
        database="pytest_test"
    )
 
    yield conn
 
    await conn.close()

# tests/test_postgres.py
import pytest
 
@pytest.mark.asyncio
async def test_postgres_connection(db_connection):
    """Проверяем что БД доступна"""
    result = await db_connection.fetchval("SELECT 1")
    assert result == 1
 
@pytest.mark.asyncio
async def test_insert_user(db_connection):
    """Вставляем пользователя"""
    await db_connection.execute(
        "INSERT INTO users (email) VALUES ($1)",
        "test@example.com"
    )
 
    count = await db_connection.fetchval("SELECT COUNT(*) FROM users")
    assert count > 0

pytest tests/test_postgres.py -v

asyncpg.UniqueViolationError: duplicate key value

# conftest.py
import pytest
import redis.asyncio as aioredis
 
@pytest.fixture
async def redis_client():
    """Async Redis client"""
    client = aioredis.from_url(
        "redis://localhost:6379/15",  # DB 15 для тестов
        decode_responses=True
    )
 
    yield client
 
    await client.flushdb()  # Чистим после теста
    await client.close()

# tests/test_redis.py
import pytest
 
@pytest.mark.asyncio
async def test_redis_set_get(redis_client):
    """Set/Get в Redis"""
    await redis_client.set("key", "value")
    result = await redis_client.get("key")
    assert result == "value"
 
@pytest.mark.asyncio
async def test_redis_isolation(redis_client):
    """Проверяем изоляцию"""
    # Ключ из предыдущего теста не должен быть виден
    result = await redis_client.get("key")
    assert result is None  # ✅ flushdb сработал

# ❌ БЕЗ pytest.ini
@pytest.mark.asyncio  # Нужен декоратор!
async def test():
    pass

# ✅ С pytest.ini
[pytest]
asyncio_mode = auto
# Декоратор не нужен для async def test_*

# ❌ ПЛОХО
asyncio_mode = strict  # Старый режим
 
# ✅ ХОРОШО
asyncio_mode = auto  # Современный

# Проверка
docker ps
 
# Если контейнеров нет:
docker start pytest-postgres pytest-redis

# Проверка портов
lsof -i :5432  # PostgreSQL
lsof -i :6379  # Redis
 
# Если порт занят, остановите процесс или измените порт
docker run -p 5433:5432 postgres:15-alpine

# tests/test_environment.py
import pytest
import asyncpg
import redis.asyncio as aioredis
 
@pytest.mark.asyncio
async def test_full_environment():
    """Проверяем что всё работает"""
    # PostgreSQL
    pg_conn = await asyncpg.connect(
        "postgresql://postgres:testpass@localhost:5432/pytest_test"
    )
    pg_result = await pg_conn.fetchval("SELECT 1")
    await pg_conn.close()
    assert pg_result == 1
 
    # Redis
    redis = aioredis.from_url("redis://localhost:6379/15")
    await redis.set("test", "ok")
    redis_result = await redis.get("test")
    await redis.close()
    assert redis_result == "ok"
 
    print("✅ Environment ready!")

pytest tests/test_environment.py -v -s

test_full_environment PASSED
✅ Environment ready!