Testando com Monkey Patching

O Cenário

Todo desenvolvedor já passou por isso: você precisa alterar ou dar manutenção em um trecho de código que não foi escrito pensando em testes. Frequentemente, esse código mistura lógica de negócio com configurações globais ou dependências implícitas, tornando a criação de testes unitários um desafio.

Um exemplo clássico, especialmente em pipelines de dados, é uma função que utiliza uma sessão Spark (spark) que existe como uma variável global no ambiente de produção, mas que não está definida no escopo de um teste local.

A solução ideal seria refatorar o código para usar Injeção de Dependência, mas nem sempre temos tempo ou permissão para fazer grandes alterações na base de código. Então, como criamos uma rede de segurança para garantir que nossas alterações funcionem? A resposta tática é Monkey Patching.

Definições Rápidas

Antes de prosseguir, vamos alinhar dois conceitos-chave:

• Injeção de Dependência (Dependency Injection - DI): Um padrão de projeto onde as dependências de um componente (objetos, configurações, conexões) são fornecidas a ele externamente, em vez de serem criadas internamente. Na prática, significa "passar o que a função precisa como parâmetro".
• Monkey Patching: Uma técnica que permite modificar ou substituir dinamicamente o comportamento de módulos, classes ou funções em tempo de execução. Em testes, usamos isso para substituir dependências reais (como bancos de dados ou APIs) por objetos falsos ("mocks").

O Código-Alvo

Imagine a seguinte função em um arquivo data_processor.py. Ela recebe um RDD, mas depende de um objeto spark que não está em sua assinatura.

# my_project/data_processor.py

from pyspark.sql import RDD, DataFrame
from pyspark.sql.functions import col, explode, split
import re

def process_raw_logs(log_rdd: RDD) -> DataFrame:
    """
    Transforma um RDD de logs brutos, quebrando JSONs concatenados
    e retornando um DataFrame estruturado.
    """
    # Mapeia o RDD para adicionar delimitadores
    mapped_rdd = log_rdd.map(
        lambda line: {"content": re.sub(r'\}\{', "}##!!##{", line)}
    )

    # CRASH! A variável 'spark' não existe no escopo do teste.
    df = spark.createDataFrame(mapped_rdd)

    df_processed = df.withColumn(
        "content", explode(split(col("content"), "##!!##"))
    )

    return df_processed

O Desafio do Teste

Um teste direto para essa função falharia, pois o spark não está definido.

# tests/test_data_processor.py

def test_process_raw_logs_fails(spark): # 'spark' aqui é uma fixture do pytest
    from my_project.data_processor import process_raw_logs

    # ... código para criar um RDD de teste ...

    # A linha abaixo irá falhar com: NameError: name 'spark' is not defined
    result = process_raw_logs(test_rdd)

A Solução: Monkey Patching em Ação

Para contornar isso, vamos usar o monkeypatch do pytest para injetar a sessão spark (fornecida pela nossa fixture de teste) diretamente no módulo data_processor antes de chamar a função.

# tests/test_data_processor.py

from pyspark.sql import SparkSession

def test_process_raw_logs_with_monkeypatch(spark: SparkSession):
    # 1. Importamos o módulo que queremos testar, não a função diretamente.
    #    Isso nos dá um objeto para "remendar".
    from my_project import data_processor

    # 2. AQUI ESTÁ O TRUQUE: Monkey Patching.
    #    Criamos um atributo chamado 'spark' dentro do módulo 'data_processor'
    #    e atribuímos a ele a nossa fixture 'spark' do teste.
    data_processor.spark = spark

    # 3. Agora, preparamos nosso cenário de teste.
    log_data = ['{"id": 1}{"id": 2}', '{"id": 3}']
    test_rdd = spark.sparkContext.parallelize(log_data)

    # 4. Executamos a função.
    #    Quando a função `process_raw_logs` procurar por 'spark', ela o encontrará
    #    no escopo do seu próprio módulo, pois nós o colocamos lá.
    result_df = data_processor.process_raw_logs(test_rdd)

    # 5. Verificamos o resultado.
    assert result_df.count() == 3
    assert "content" in result_df.columns

    # O pytest garante que essa modificação no módulo seja desfeita após o teste,
    # evitando contaminação entre testes.

Por Que Isso Funciona?

Em Python, módulos são objetos. Quando você faz import my_project.data_processor, você está carregando o código daquele arquivo em um objeto de módulo na memória. O que a linha data_processor.spark = spark faz é simplesmente adicionar um novo atributo a esse objeto. A função process_raw_logs, ao ser executada, resolve o nome spark procurando primeiro em seu escopo local e depois no escopo do módulo onde foi definida, encontrando a nossa versão injetada.

Análise da Abordagem

Esta técnica é poderosa, mas deve ser usada com cautela.

• Prós:

  • Permite testar código que de outra forma seria "intestável".
  • Serve como uma "rede de segurança" essencial para permitir futuras refatorações. Você pode escrever um teste como este para garantir que não quebrou nada ao fazer uma alteração.

• Contras:

  • É um "code smell". O teste se torna mais complexo e menos legível.
  • A dependência da função continua oculta. Um desenvolvedor precisa ler o teste para entender que a função process_raw_logs depende de spark.
  • O teste fica fortemente acoplado à estrutura do arquivo, não ao contrato da função.

Conclusão

Enfrentar código legado ou não projetado para testes é uma realidade. Embora a Injeção de Dependência seja o objetivo estratégico para um código limpo e manutenível, o Monkey Patching é uma ferramenta tática indispensável. Ele nos permite criar uma rede de segurança, garantindo a qualidade e a estabilidade do software enquanto pavimentamos o caminho para futuras melhorias. Use-o como um meio para um fim, não como o padrão final.

---

Referências

• Pytest monkeypatch fixture: pytest.org documentation
• unittest.mock (Standard Library): docs.python.org
• Dependency Injection - Martin Fowler: martinfowler.com
• Working Effectively with Legacy Code (Livro de Michael Feathers): Um livro fundamental sobre as estratégias discutidas aqui, como "Characterization Tests" e "Seams".