11 - Testing Profesional con Pytest

1. De Vitest/Jest a Pytest

// Vitest/Jest
import { describe, it, expect, beforeEach } from 'vitest';

describe('UserService', () => {
let db: Database;

beforeEach(async () => {
  db = await createTestDb();
});

afterEach(async () => {
  await db.cleanup();
});

it('should create user', async () => {
  const user = await userService.create({ email: 'test@test.com' });
  expect(user.id).toBeDefined();
});
});

# Pytest
import pytest

@pytest.fixture
async def db():
  connection = await create_test_db()
  yield connection  # Test runs here
  await connection.cleanup()  # Teardown

async def test_create_user(db):  # Fixture inyectada por nombre
  user = await user_service.create(db, email="test@test.com")
  assert user.id is not None

Diferencias clave:

• Fixtures vs Hooks: En Pytest, las fixtures son funciones inyectadas por nombre
• Yield: Setup antes del yield, teardown después
• Scopes: Las fixtures tienen alcances (function, class, module, session)
• Sin describe: Los tests se agrupan por archivo/clase

2. Configuración de Pytest

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Ejecuta tests async automáticamente
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_functions = ["test_*"]
addopts = [
    "-v",                    # Verbose
    "--tb=short",            # Traceback corto
    "-x",                    # Stop on first failure
    "--strict-markers",      # Error on unknown markers
    "-p", "no:warnings",     # Disable warnings
]
markers = [
    "slow: marks tests as slow",
    "integration: marks tests as integration tests",
]
filterwarnings = [
    "ignore::DeprecationWarning",
]

Estructura de Tests

tests/
├── conftest.py              # Fixtures compartidas
├── unit/
│   ├── conftest.py          # Fixtures de unit tests
│   ├── test_services.py
│   └── test_utils.py
├── integration/
│   ├── conftest.py          # Fixtures de integration
│   ├── test_api.py
│   └── test_database.py
└── e2e/
    └── test_flows.py

3. Fixtures: El Corazón de Pytest

Fixture Básica

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

from app.main import app
from app.database import Base, get_db
from app.core.config import settings

# Engine de test
TEST_DATABASE_URL = "postgresql+asyncpg://test:test@localhost:5432/test_db"

@pytest.fixture(scope="session")
def event_loop():
    """Create event loop for session-scoped async fixtures."""
    import asyncio
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def engine():
    """Create test database engine."""
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    yield engine
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    
    await engine.dispose()

@pytest.fixture
async def db(engine) -> AsyncGenerator[AsyncSession, None]:
    """Create a fresh database session for each test."""
    async with AsyncSession(engine, expire_on_commit=False) as session:
        yield session
        await session.rollback()  # Rollback after each test

Fixture de Cliente HTTP

@pytest.fixture
async def client(db: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """HTTP client with database override."""
    
    async def override_get_db():
        yield db
    
    app.dependency_overrides[get_db] = override_get_db
    
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac
    
    app.dependency_overrides.clear()

Fixture de Usuario Autenticado

@pytest.fixture
async def test_user(db: AsyncSession) -> User:
    """Create a test user."""
    user = User(
        email="test@example.com",
        hashed_password=hash_password("testpassword123"),
        is_active=True,
    )
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user

@pytest.fixture
async def auth_headers(test_user: User) -> dict[str, str]:
    """Generate auth headers for test user."""
    token = create_access_token(str(test_user.id), roles=[], scopes=[])
    return {"Authorization": f"Bearer {token}"}

@pytest.fixture
async def authenticated_client(
    client: AsyncClient,
    auth_headers: dict[str, str],
) -> AsyncClient:
    """Client with authentication headers."""
    client.headers.update(auth_headers)
    return client

4. Scopes de Fixtures

@pytest.fixture(scope="session")
async def database_engine():
    """Expensive: create once per test session."""
    engine = create_async_engine(...)
    yield engine
    await engine.dispose()

@pytest.fixture(scope="module")
async def seeded_data(database_engine):
    """Seed data once per test file."""
    await seed_test_data(database_engine)
    yield
    await cleanup_test_data(database_engine)

@pytest.fixture  # scope="function" by default
async def db_session(database_engine):
    """Fresh session for each test."""
    async with AsyncSession(database_engine) as session:
        yield session
        await session.rollback()

5. Testing de Endpoints

Test Básico de API

# tests/integration/test_users_api.py
import pytest
from httpx import AsyncClient

async def test_create_user(client: AsyncClient):
    response = await client.post(
        "/users",
        json={"email": "new@example.com", "password": "SecurePass123!"},
    )
    
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "new@example.com"
    assert "id" in data
    assert "password" not in data  # Sensible field excluded

async def test_create_user_duplicate_email(client: AsyncClient, test_user: User):
    response = await client.post(
        "/users",
        json={"email": test_user.email, "password": "AnotherPass123!"},
    )
    
    assert response.status_code == 409
    assert response.json()["error"]["code"] == "CONFLICT"

async def test_get_user_unauthorized(client: AsyncClient):
    response = await client.get("/users/me")
    
    assert response.status_code == 401

async def test_get_user_authenticated(
    authenticated_client: AsyncClient,
    test_user: User,
):
    response = await authenticated_client.get("/users/me")
    
    assert response.status_code == 200
    assert response.json()["id"] == test_user.id

Parametrización

@pytest.mark.parametrize("email,expected_status", [
    ("valid@email.com", 201),
    ("invalid-email", 422),
    ("", 422),
    ("a" * 300 + "@test.com", 422),  # Too long
])
async def test_create_user_email_validation(
    client: AsyncClient,
    email: str,
    expected_status: int,
):
    response = await client.post(
        "/users",
        json={"email": email, "password": "ValidPass123!"},
    )
    assert response.status_code == expected_status

@pytest.mark.parametrize("password,valid", [
    ("short", False),
    ("nouppercase123!", False),
    ("NOLOWERCASE123!", False),
    ("NoNumbers!!", False),
    ("NoSpecialChar123", False),
    ("ValidPassword123!", True),
])
async def test_password_validation(password: str, valid: bool):
    from app.schemas.auth import PasswordPolicy
    
    if valid:
        PasswordPolicy(password=password)  # Should not raise
    else:
        with pytest.raises(ValueError):
            PasswordPolicy(password=password)

6. Mocking y Patching

Monkeypatch (Built-in)

async def test_external_api_failure(client: AsyncClient, monkeypatch):
    """Test behavior when external API fails."""
    
    async def mock_fetch(*args, **kwargs):
        raise httpx.HTTPError("Connection failed")
    
    monkeypatch.setattr("app.services.external.fetch_data", mock_fetch)
    
    response = await client.get("/data/external")
    assert response.status_code == 503
    assert "external service unavailable" in response.json()["error"]["message"].lower()

unittest.mock con AsyncMock

from unittest.mock import AsyncMock, patch, MagicMock

async def test_email_sent_on_registration(client: AsyncClient):
    with patch("app.services.email.send_email", new_callable=AsyncMock) as mock_send:
        mock_send.return_value = {"status": "sent"}
        
        response = await client.post(
            "/users",
            json={"email": "new@example.com", "password": "SecurePass123!"},
        )
        
        assert response.status_code == 201
        mock_send.assert_called_once()
        call_args = mock_send.call_args
        assert call_args[1]["to"] == "new@example.com"
        assert "welcome" in call_args[1]["template"].lower()

Dependency Override (Preferred for FastAPI)

async def test_with_mock_cache(client: AsyncClient):
    """Override Redis dependency with mock."""
    
    mock_cache = AsyncMock()
    mock_cache.get.return_value = None
    mock_cache.set.return_value = True
    
    async def get_mock_cache():
        return mock_cache
    
    app.dependency_overrides[get_cache] = get_mock_cache
    
    response = await client.get("/cached-data")
    
    assert response.status_code == 200
    mock_cache.get.assert_called()
    
    app.dependency_overrides.clear()

7. Testcontainers: Bases de Datos Reales

# tests/conftest.py
import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres_container():
    """Spin up a real PostgreSQL container."""
    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres

@pytest.fixture(scope="session")
async def engine(postgres_container):
    """Create engine connected to container."""
    url = postgres_container.get_connection_url()
    # Convert to async URL
    async_url = url.replace("postgresql://", "postgresql+asyncpg://")
    
    engine = create_async_engine(async_url)
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    yield engine
    
    await engine.dispose()

Con Redis

from testcontainers.redis import RedisContainer

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

@pytest.fixture
async def redis_client(redis_container):
    import redis.asyncio as aioredis
    
    client = aioredis.from_url(
        f"redis://{redis_container.get_container_host_ip()}:{redis_container.get_exposed_port(6379)}"
    )
    yield client
    await client.close()

8. Testing de Servicios (Unit Tests)

# tests/unit/test_user_service.py
import pytest
from unittest.mock import AsyncMock, MagicMock

from app.services.user import UserService
from app.exceptions import NotFoundError, ConflictError

@pytest.fixture
def mock_repo():
    return AsyncMock()

@pytest.fixture
def mock_cache():
    return AsyncMock()

@pytest.fixture
def user_service(mock_repo, mock_cache):
    return UserService(repo=mock_repo, cache=mock_cache)

async def test_get_user_from_cache(user_service, mock_repo, mock_cache):
    """Test that cached user is returned without DB hit."""
    cached_user = {"id": 1, "email": "cached@example.com"}
    mock_cache.get.return_value = cached_user
    
    result = await user_service.get_user(1)
    
    assert result == cached_user
    mock_cache.get.assert_called_once_with("user:1")
    mock_repo.get.assert_not_called()  # No DB hit

async def test_get_user_cache_miss(user_service, mock_repo, mock_cache):
    """Test that user is fetched from DB on cache miss."""
    mock_cache.get.return_value = None
    db_user = MagicMock(id=1, email="db@example.com")
    mock_repo.get.return_value = db_user
    
    result = await user_service.get_user(1)
    
    assert result.id == 1
    mock_cache.set.assert_called_once()  # Cache updated

async def test_get_user_not_found(user_service, mock_repo, mock_cache):
    """Test NotFoundError is raised for non-existent user."""
    mock_cache.get.return_value = None
    mock_repo.get.return_value = None
    
    with pytest.raises(NotFoundError) as exc_info:
        await user_service.get_user(999)
    
    assert "999" in str(exc_info.value)

async def test_create_user_duplicate_email(user_service, mock_repo):
    """Test ConflictError on duplicate email."""
    mock_repo.exists_by_email.return_value = True
    
    with pytest.raises(ConflictError):
        await user_service.create_user({"email": "exists@example.com"})

9. Markers y Selección de Tests

# tests/integration/test_slow.py
import pytest

@pytest.mark.slow
async def test_complex_report_generation(client: AsyncClient):
    """This test takes a long time."""
    response = await client.post("/reports/generate", json={"type": "full"})
    assert response.status_code == 200

@pytest.mark.integration
async def test_external_api_integration(client: AsyncClient):
    """Requires external service."""
    pass

@pytest.mark.skip(reason="Feature not implemented yet")
async def test_future_feature():
    pass

@pytest.mark.skipif(
    os.getenv("CI") == "true",
    reason="Skip in CI environment"
)
async def test_local_only():
    pass

# Ejecutar solo tests rápidos
pytest -m "not slow"

# Ejecutar solo integration tests
pytest -m integration

# Ejecutar todo excepto integration
pytest -m "not integration"

# Combinar markers
pytest -m "integration and not slow"

10. Coverage y CI

pytest-cov

# Ejecutar con coverage
pytest --cov=app --cov-report=html --cov-report=term-missing

# Con umbral mínimo
pytest --cov=app --cov-fail-under=80

GitHub Actions

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: test_db
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install uv
        run: pip install uv
      
      - name: Install dependencies
        run: uv sync --frozen
      
      - name: Run tests
        env:
          DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/test_db
        run: |
          uv run pytest --cov=app --cov-report=xml --cov-fail-under=80
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml

11. Tabla Comparativa

Conclusión

Pytest ofrece un modelo de testing más flexible que Jest/Vitest:

1. Fixtures como DI — Inyección por nombre, scopes granulares
2. Yield pattern — Setup y teardown en la misma función
3. Dependency overrides — Testing de FastAPI sin mocks complejos
4. Testcontainers — Bases de datos reales en tests

Pattern Senior: Usa fixtures de scope session para recursos costosos (DB engine), scope function para aislamiento (sessions), y dependency_overrides para mockear servicios externos.

En el siguiente capítulo, implementaremos logging estructurado y observabilidad.