Gerador de CPF em pytest

conftest.py compartilhado

# tests/conftest.py
import pytest
from fakeforge import FakeForge

@pytest.fixture(scope="session")
def ff():
    """SDK instance compartilhada entre TODOS os testes da sessão."""
    return FakeForge()

@pytest.fixture(scope="session")
def cpfs_batch(ff):
    """500 CPFs pré-carregados. Reusa entre testes."""
    return ff.cpf(500)

@pytest.fixture(scope="function")
def customer(ff):
    """1 customer novo por teste. Correlacionado."""
    return ff.preset("customer", 1)[0]

@pytest.fixture(scope="session")
def fintech_customers(ff):
    """50 clientes fintech com PIX + banco + cartão + score."""
    return ff.preset("fintech", 50)

Teste de signup (parametrize)

# tests/test_signup.py
import pytest
from myapp.services import SignupService

class TestSignup:
    def test_aceita_100_cpfs_validos(self, cpfs_batch):
        service = SignupService()
        for cpf in cpfs_batch[:100]:
            resultado = service.criar_conta(cpf=cpf, email=f"user-{cpf[-2:]}@test.com")
            assert resultado.status == "created"

    def test_rejeita_cpf_invalido(self):
        service = SignupService()
        with pytest.raises(ValueError, match="CPF inválido"):
            service.criar_conta(cpf="111.111.111-11", email="user@test.com")

    @pytest.mark.parametrize("indice", range(10))
    def test_customer_completo(self, ff, indice):
        [c] = ff.preset("customer", 1)
        assert c["cpf"]
        assert c["email"].endswith("@gmail.com") or c["email"].endswith("@outlook.com")
        assert len(c["telefone"]) >= 14

Django tests com pytest-django

# tests/test_customer_model.py
import pytest
from myapp.models import Customer

@pytest.mark.django_db
class TestCustomerModel:
    def test_salvar_customer_correlacionado(self, customer):
        obj = Customer.objects.create(
            nome=customer["nome"],
            cpf=customer["cpf"],
            email=customer["email"],
            telefone=customer["telefone"],
        )
        assert obj.pk is not None
        # Email tem que derivar do nome
        primeiro_nome = customer["nome"].split()[0].lower()
        assert primeiro_nome in customer["email"].lower()

    def test_bulk_seed_1000_customers(self, ff):
        dados = ff.preset("customer", 1000)
        Customer.objects.bulk_create([
            Customer(**{k: d[k] for k in ("nome", "cpf", "email", "telefone")})
            for d in dados
        ])
        assert Customer.objects.count() == 1000

Teste de fintech (preset fintech)

# tests/test_credit_engine.py
import pytest
from myapp.credit import CreditEngine

class TestCreditEngine:
    def test_score_distribution_realista(self, fintech_customers):
        """Score Serasa segue distribuição normal ponderada (média 650)."""
        engine = CreditEngine()
        aprovados = 0
        rejeitados = 0

        for c in fintech_customers:
            resultado = engine.avaliar(
                cpf=c["customer"]["cpf"],
                score=c["customer"]["score_serasa"],
                renda=c["customer"]["renda_mensal"],
            )
            if resultado.aprovado:
                aprovados += 1
            else:
                rejeitados += 1

        # ~40% aprovação esperada em base realista
        taxa = aprovados / len(fintech_customers)
        assert 0.30 <= taxa <= 0.55

Mock com pytest-mock

# tests/test_kyc.py
def test_kyc_com_customer_fake(mocker, customer):
    mock_serasa = mocker.patch("myapp.serasa.consultar_score")
    mock_serasa.return_value = 750

    from myapp.kyc import KycService
    resultado = KycService().validar(cpf=customer["cpf"], nome=customer["nome"])

    assert resultado.status == "aprovado"
    mock_serasa.assert_called_once_with(customer["cpf"])

Rodando com paralelismo (pytest-xdist)

# pytest -n auto (roda em N processos)
# Cuidado: cada worker chama a API 1 vez pro session fixture
# Free tier 50/dia pode estourar com xdist + muitos testes

# Solução: cache em disco
@pytest.fixture(scope="session")
def cpfs_batch(ff, tmp_path_factory, worker_id):
    if worker_id == "master":
        return ff.cpf(500)
    # workers reusa cache do master (via arquivo temp compartilhado)
    cache_file = tmp_path_factory.getbasetemp().parent / "cpfs.json"
    if cache_file.exists():
        import json
        return json.loads(cache_file.read_text())
    data = ff.cpf(500)
    cache_file.write_text(json.dumps(data))
    return data

Próximos passos