Testes¶

Filosofia, tipos, como rodar e como escrever cada camada.

A pirâmide¶

        ▲
       /E2E\        poucos — contratos HTTP críticos
      /──────\
     /integrat\     médios — repositórios contra banco real
    /──────────\
   /    unit    \   muitos — lógica de domínio e UCs
  ──────────────────

Unit é a base: rápido, sem I/O, roda em qualquer máquina sem setup. Qualquer lógica de domínio ou UC que pode ser testada com mocks deve ser testada aqui.

Integration valida que o SQL dos repositórios funciona contra o PostgreSQL real — esquemas, constraints, índices, queries complexas. Não testa lógica de negócio (isso é responsabilidade do unit).

E2E valida os contratos HTTP das surfaces — status codes, estrutura de resposta, autenticação, fluxos críticos end-to-end. Não valida cada variação de regra de negócio (isso é integration + unit).

Tipos de teste¶

Unit (`tests/unit/`)¶

Quando usar: lógica de domínio, Use Cases, Services, mappers, validators — qualquer coisa que não precisa de banco ou HTTP.

O que não fazer: não crie mocks de repositório para testar que "save foi chamado". Teste o comportamento do UC — o resultado retornado, as exceções levantadas, o estado das entidades criadas.

Marcador: @pytest.mark.unit (não é necessário adicionar manualmente — basta estar em tests/unit/)

Setup: nenhum. Roda direto sem banco ou Redis.

Exemplo — UC com repositório mockado:

import pytest
from unittest.mock import AsyncMock, MagicMock
from contextlib import asynccontextmanager

from src.modules.billing.application.subscription.ucs.create_checkout_session import (
    CreateCheckoutSession, CreateCheckoutSessionCommand,
)

def _make_uow() -> AsyncMock:
    """Mock UoW que cede uma tx mock ao UC."""
    tx = MagicMock()
    uow = AsyncMock()

    @asynccontextmanager
    async def _begin():
        yield tx

    uow.begin = _begin
    return uow

class TestCreateCheckoutSession:
    async def test_execute_creates_checkout_for_valid_stripe_plan(self) -> None:
        plan_reader = AsyncMock()
        plan_reader.require_active_version_by_id.return_value = _create_stripe_plan_version()

        subscription_repo = AsyncMock()
        subscription_repo.get_active_by_tenant.return_value = None

        payment_provider = AsyncMock()
        payment_provider.create_checkout_session.return_value = "https://checkout.stripe.com/x"

        uc = CreateCheckoutSession(
            uow=_make_uow(),
            plan_reader=plan_reader,
            subscription_repository=subscription_repo,
            payment_provider=payment_provider,
        )

        result = await uc.execute(CreateCheckoutSessionCommand(
            tenant_id=TenantID.generate(),
            plan_version_id=PlanVersionID.generate(),
            success_url="https://example.com/success",
            cancel_url="https://example.com/cancel",
        ))

        assert result.checkout_url == "https://checkout.stripe.com/x"

Builder functions para dados de teste — use funções modulares no topo do arquivo, não factories globais:

def _create_stripe_plan_version(
    plan_version_id: PlanVersionID | None = None,
    stripe_price_id: str = "price_123",
) -> PlanVersion:
    return PlanVersion(
        id=plan_version_id or PlanVersionID.generate(),
        plan_id=PlanID.generate(),
        currency=BillingCurrency.USD,
        price_monthly_cents=9900,
        stripe_price_id_monthly=stripe_price_id,
    )

Prefira builders simples (funções _create_*) ao invés de fixtures pytest para dados de domínio — mais legível e sem acoplamento ao ciclo de vida do pytest.

Integration (`tests/integration/`)¶

Quando usar: repositórios (SQL real), queries complexas, constraints de banco, uso de pgvector.

O que não fazer: não teste lógica de negócio aqui. Se o teste não precisa do banco, é um unit test.

Marcador: adicionado automaticamente a todos os arquivos em tests/integration/ via pytest_collection_modifyitems.

Setup necessário:

make db-start    # PostgreSQL local via Supabase CLI
make db-reset    # aplica migrations + seeds de teste

Se o banco não estiver rodando, o fixture db_engine chama pytest.skip() automaticamente — o teste pula com mensagem explicativa em vez de falhar com erro obscuro.

Isolamento via savepoints¶

Cada teste de integração roda dentro de uma transação que é revertida ao final — sem precisar limpar os dados manualmente. O mecanismo usa savepoints (nested transactions) do PostgreSQL:

db_engine (session)
  └─ db_connection (function) — begin outer transaction
       └─ db_session (function) — nested transaction (savepoint)
            └─ tx (function) — PgTransaction wrapping db_session
                 └─ [test runs]
            └─ savepoint is rolled back
       └─ outer transaction is rolled back

Isso significa que múltiplos commit() dentro de um teste funcionam normalmente — cada um promove o savepoint —, mas tudo é desfeito ao final do teste.

Fixtures disponíveis¶

Fixture	Escopo	Tipo	O que fornece
`db_engine`	session	`AsyncEngine`	Engine conectado ao banco de teste
`db_connection`	function	`AsyncConnection`	Conexão com transação aberta
`db_session`	function	`AsyncSession`	Session com rollback automático
`tx`	function	`PgTransaction`	Wrapper de `Transaction` para repositórios
`integration_settings`	session	`IntegrationTestSettings`	Settings mínimos sem Doppler

Seed data¶

Os testes de integração usam dados pré-semeados em supabase/seeds/test_data.sql. As constantes ficam em tests/fixtures/seed_data.py:

from tests.fixtures.seed_data import TEST_TENANTS, TEST_USERS, TEST_PASSWORD

TEST_TENANTS["acme"]    # "tenant_01HQ7ZACMECORP000000000000"
TEST_TENANTS["startup"] # "tenant_01HQ7ZSTARTUPINC0000000000"

TEST_USERS["owner_acme"]  # ID do owner do tenant acme
TEST_USER_EMAILS["owner_acme"]  # "[email protected]"
TEST_PASSWORD  # "Test@123456"

Dois tenants disponíveis (acme e startup) para testar isolamento entre tenants.

Exemplo de teste de repositório:

from tests.fixtures.seed_data import TEST_TENANTS
from src.modules.shared.infrastructure.postgresql.transaction import PgTransaction

class TestContactRepository:
    async def test_save_and_get_contact(self, tx: PgTransaction) -> None:
        repo = ContactRepository()
        tenant_id = TenantID.validate(TEST_TENANTS["acme"])

        contact = Contact(
            tenant_id=tenant_id,
            name="Test Contact",
        )
        saved = await repo.save(tx, contact)
        fetched = await repo.get_by_id(tx, saved.id, tenant_id)

        assert fetched is not None
        assert fetched.name == "Test Contact"
        # não precisa limpar — rollback automático

E2E (`tests/e2e/`)¶

Quando usar: contratos HTTP das surfaces (status codes, estrutura de resposta, autenticação, fluxos end-to-end críticos como login → criar recurso → verificar estado).

O que não fazer: não cubra cada variação de regra de negócio aqui — o custo de manutenção é alto. Um e2e por fluxo crítico, não por regra de negócio.

Marcador: @pytest.mark.e2e (adicionado automaticamente a todos em tests/e2e/)

Setup necessário: Doppler configurado + banco rodando (ou Supabase Preview no CI).

Clients disponíveis¶

Fixture	Escopo	Quando usar
`simple_async_client`	session	Endpoints que usam a conexão própria do app (login, health)
`async_test_client`	function	Endpoints tenant-scoped com override da transação de teste

async_test_client sobrescreve _get_transaction com a sessão de teste — os dados criados pelo endpoint são visíveis para asserção mas revertidos ao final do teste.

Helper de autenticação¶

from tests.fixtures.api_client import async_login
from tests.fixtures.seed_data import TEST_USER_EMAILS, TEST_TENANTS

token = await async_login(
    client,
    email=TEST_USER_EMAILS["owner_acme"],
    tenant_id=TEST_TENANTS["acme"],
)
# tokens são cacheados por (email, tenant_id) para não repetir login entre testes

Exemplo de teste e2e:

class TestLoginEndpoint:
    async def test_login_with_valid_credentials_returns_tokens(
        self, simple_async_client: AsyncClient
    ) -> None:
        response = await simple_async_client.post(
            "/v1/auth/login",
            json={
                "grant_type": "password",
                "email": TEST_USER_EMAILS["owner_acme"],
                "password": TEST_PASSWORD,
            },
        )

        assert response.status_code == status.HTTP_200_OK, response.text
        data = response.json()
        assert "access_token" in data
        assert data["token_type"] == "Bearer"

Como rodar¶

Comando	O que roda	Precisa de banco?
`make test-unit`	Apenas `tests/unit/`	Não
`make test-integration`	Apenas `tests/integration/`	Sim (local)
`make test-e2e`	Apenas `tests/e2e/`	Sim (local + Doppler)
`make test`	Unit + E2E (sem integration)	Não (para unit)
`make test-all`	Tudo	Sim

Rodando testes específicos¶

# Um arquivo
uv run pytest tests/unit/billing/test_create_checkout_session.py -v

# Uma classe
uv run pytest tests/unit/billing/test_create_checkout_session.py::TestCreateCheckoutSession -v

# Um teste
uv run pytest tests/unit/billing/test_create_checkout_session.py::TestCreateCheckoutSession::test_execute_creates_checkout_for_valid_stripe_plan -v

# Por módulo (todos os testes do billing)
uv run pytest tests/ -k "billing" -v

# Só testes que falharam na última execução
uv run pytest tests/unit --lf -v

Parallelismo¶

Os testes de integration e e2e no CI rodam com pytest-xdist:

uv run pytest tests/integration -n auto   # auto-detecta CPUs
uv run pytest tests/e2e -n 4              # fixo em 4 workers

Localmente não é necessário usar -n — a paralelização é mais útil no CI onde os testes de integração contra o Supabase Preview são lentos.

Cobertura¶

Não há threshold de cobertura configurado — pytest-cov está instalado mas o gatilho é qualitativo: código crítico de domínio e UCs devem ter unit tests. Rotas de erro, validações e fluxos alternativos têm mais valor que número de linhas cobertas.

Para gerar um relatório de cobertura:

uv run pytest tests/unit --cov=src --cov-report=html
open htmlcov/index.html

Isolamento e boas práticas¶

Não compartilhe estado entre testes¶

Cada teste deve ser independente. Nunca dependa da ordem de execução ou de dados deixados por outro teste.

Unit: sem estado compartilhado por definição (sem banco)
Integration: rollback automático garante isolamento por função
E2E com async_test_client: rollback automático
E2E com simple_async_client: dados persistem — use IDs únicos por teste ou verifique estado sem depender de ausência de dados de outros testes

Mock de side effects¶

O fixture mock_send_email é ativado automaticamente para todos os testes de integration. Isso evita que testes de repositório disparem emails reais acidentalmente. O mesmo padrão vale para outros side effects (WAHA, webhooks):

# mock_send_email é um FakeSendEmail que captura chamadas
async def test_something_that_sends_email(
    mock_send_email: FakeSendEmail,
    tx: PgTransaction,
) -> None:
    await uc.execute(command)
    assert len(mock_send_email.sent) == 1
    assert mock_send_email.sent[0].to == "[email protected]"

TDD em bug fixes¶

Bug fixes devem começar com um teste que reproduz o bug. O fluxo:

Escreva o teste que falha (reproduz o bug)
Confirme que falha pelo motivo certo
Corrija o código
Confirme que o teste passa
Verifique que nada mais quebrou

Isso garante que o bug não volta sem ser detectado.