Biblioteca Snowpark Checkpoints: hipótese¶

Teste de unidade do Hypothesis¶

Hypothesis é uma biblioteca de testes avançada para Python, projetada para aprimorar os testes de unidade tradicionais, gerando automaticamente uma ampla variedade de dados de entrada. Ele usa testes baseados em propriedades, nos quais, em vez de especificar casos de teste individuais, você pode descrever o comportamento esperado do seu código com propriedades ou condições, e o Hypothesis gera exemplos para testar essas propriedades minuciosamente. Essa abordagem ajuda a descobrir casos extremos e comportamentos inesperados, o que a torna especialmente eficaz para funções complexas. Para obter mais informações, consulte Hypothesis.

O pacote snowpark-checkpoints-hypothesis amplia a biblioteca Hypothesis para gerar Snowpark DataFrames sintéticos para fins de teste. Ao aproveitar a capacidade do Hypothesis de gerar dados de teste diversos e aleatórios, você pode criar o Snowpark DataFrames com esquemas e valores variados para simular cenários do mundo real e descobrir casos extremos, garantindo um código robusto e verificando a correção de transformações complexas.

A estratégia do Hypothesis para o Snowpark conta com o Pandera para gerar dados sintéticos. A função dataframe_strategy usa o esquema especificado para gerar um Pandas DataFrame que esteja em conformidade com ele e, em seguida, converte-o em um Snowpark DataFrame.

Assinatura da função:

def dataframe_strategy(
  schema: Union[str, DataFrameSchema],
  session: Session,
  size: Optional[int] = None
) -> SearchStrategy[DataFrame]

Parâmetros de função:

• schema: o esquema que define as colunas, os tipos de dados e as verificações que Snowpark dataframe gerado deve corresponder. O esquema pode ser:

  • Um caminho para um arquivo de esquema JSON gerado pela função collect_dataframe_checkpoint do pacote snowpark-checkpoints-collectors.

  • Uma instância de pandera.api.pandas.container.DataFrameSchema.

• session: uma instância de snowflake.snowpark.Session que será usada para criar Snowpark DataFrames.

• size: número de linhas a serem geradas para cada Snowpark DataFrame. Se esse parâmetro não for fornecido, a estratégia gerará DataFrames de tamanhos diferentes.

Saída de função:

Retorna uma SearchStrategy do Hypothesis que gera o Snowpark DataFrames.

Tipos de dados suportados e não suportados¶

A função dataframe_strategy oferece suporte à geração de Snowpark DataFrames com diferentes tipos de dados. Dependendo do tipo do argumento do esquema passado para a função, os tipos de dados suportados pela estratégia variam. Observe que, se a estratégia encontrar um tipo de dados não suportado, ela abrirá uma exceção.

A tabela a seguir mostra os tipos de dados PySpark suportados e não suportados pela função dataframe_strategy ao passar um arquivo JSON como argumento schema.

A tabela a seguir mostra os tipos de dados do Pandera suportados pela função dataframe_strategy ao passar um objeto DataFrameSchema como argumento schema e os tipos de dados do Snowpark para os quais eles são mapeados.

Exemplos¶

O fluxo de trabalho típico para usar a biblioteca Hypothesis para gerar o Snowpark DataFrames é o seguinte:

1. Crie uma função de teste Python padrão com as diferentes asserções ou condições que seu código deve satisfazer para todas as entradas.

2. Adicione o decorador Hypothesis @given à sua função de teste e passe a função dataframe_strategy como argumento. Para obter mais informações sobre o decorador @given, consulte hypothesis.given.

3. Execute a função de teste. Quando o teste for executado, o Hypothesis fornecerá automaticamente as entradas geradas como argumentos para o teste.

Exemplo 1: gerar Snowpark DataFrames a partir de um arquivo JSON

Veja abaixo um exemplo de como gerar Snowpark DataFrames a partir de um arquivo de esquema JSON gerado pela função collect_dataframe_checkpoint do pacote snowpark-checkpoints-collectors.

from hypothesis import given

from snowflake.hypothesis_snowpark import dataframe_strategy
from snowflake.snowpark import DataFrame, Session


@given(
    df=dataframe_strategy(
        schema="path/to/file.json",
        session=Session.builder.getOrCreate(),
        size=10,
    )
)
def test_my_function_from_json_file(df: DataFrame):
    # Test a particular function using the generated Snowpark DataFrame
    ...

Exemplo 2: gerar um Snowpark DataFrame a partir de um objeto Pandera DataFrameSchema

Abaixo está um exemplo de como gerar Snowpark DataFrames a partir de uma instância de um Pandera DataFrameSchema. Para obter mais informações, consulte Pandera DataFrameSchema.

import pandera as pa

from hypothesis import given

from snowflake.hypothesis_snowpark import dataframe_strategy
from snowflake.snowpark import DataFrame, Session


@given(
    df=dataframe_strategy(
        schema=pa.DataFrameSchema(
            {
                "boolean_column": pa.Column(bool),
                "integer_column": pa.Column("int64", pa.Check.in_range(0, 9)),
                "float_column": pa.Column(pa.Float32, pa.Check.in_range(10.5, 20.5)),
            }
        ),
        session=Session.builder.getOrCreate(),
        size=10,
    )
)
def test_my_function_from_dataframeschema_object(df: DataFrame):
    # Test a particular function using the generated Snowpark DataFrame
    ...

Exemplo 3: personalizar o comportamento do Hypothesis

Você também pode personalizar o comportamento do seu teste com o decorador Hypothesis @settings. Esse decorador permite que você personalize vários parâmetros de configuração para adequar o comportamento do teste às suas necessidades. Ao usar o decorador @settings, você pode controlar aspectos como o número máximo de casos de teste, o prazo para a execução de cada teste, os níveis de detalhamento e muitos outros. Para obter mais informações, consulte Configurações do Hypothesis.

from datetime import timedelta

from hypothesis import given, settings
from snowflake.snowpark import DataFrame, Session

from snowflake.hypothesis_snowpark import dataframe_strategy


@given(
    df=dataframe_strategy(
        schema="path/to/file.json",
        session=Session.builder.getOrCreate(),
    )
)
@settings(
    deadline=timedelta(milliseconds=800),
    max_examples=25,
)
def test_my_function(df: DataFrame):
    # Test a particular function using the generated Snowpark DataFrame
    ...