Bibliothèque de points de contrôle Snowpark : Hypothèse¶

Test unitaire Hypothesis¶

Hypothesis est une puissante bibliothèque de tests pour Python conçue pour améliorer les tests unitaires traditionnels en générant automatiquement un large éventail de données d’entrée. Elle utilise des tests basés sur les propriétés, où au lieu de spécifier des cas de test individuels, vous pouvez décrire le comportement attendu de votre code avec des propriétés ou des conditions. Hypothesis génère des exemples pour tester ces propriétés en profondeur. Cette approche permet de découvrir les cas limites et les comportements inattendus, ce qui la rend particulièrement efficace pour les fonctions complexes. Pour plus d’informations, voir Hypothesis.

Le paquet snowpark-checkpoints-hypothesis étend la bibliothèque Hypothesis pour générer des DataFrames Snowpark synthétiques à des fins de test. En tirant parti de la possibilité qu’offre Hypothesis de générer des données de test diverses et aléatoires, vous pouvez créer des DataFrames Snowpark avec des schémas et des valeurs variés pour simuler des scénarios réels et découvrir des cas limites, en garantissant la robustesse du code et en vérifiant l’exactitude des transformations complexes.

La stratégie d’Hypothesis pour Snowpark s’appuie sur Pandera pour générer des données synthétiques. La fonction dataframe_strategy utilise le schéma spécifié pour générer un DataFrame Pandas qui s’y conforme et le convertit ensuite en un DataFrame Snowpark.

Signature de la fonction :

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

Paramètres de fonction :

• schema : le schéma qui définit les colonnes, les types de données et les contrôles auxquels le dataframe Snowpark généré doit correspondre. Le schéma peut être le suivant :

  • Un chemin vers un fichier de schéma JSON généré par la fonction collect_dataframe_checkpoint du paquet snowpark-checkpoints-collectors.

  • Une instance de pandera.api.pandas.container.DataFrameSchema.

• session : une instance de snowflake.snowpark.Session qui sera utilisée pour créer les DataFrames Snowpark.

• size : le nombre de lignes à générer pour chaque DataFrame Snowpark. Si ce paramètre n’est pas fourni, la stratégie générera des DataFrames de tailles différentes.

Sortie de fonction :

Renvoie une SearchStrategy Hypothesis qui génère des DataFrames Snowpark.

Types de données pris en charge et non pris en charge¶

La fonction dataframe_strategy permet de générer les DataFrames Snowpark avec différents types de données. Selon le type de l’argument du schéma transmis à la fonction, les types de données pris en charge par la stratégie varieront. Notez que si la stratégie trouve un type de données non pris en charge, elle lèvera une exception.

Le tableau suivant indique les types de données PySpark pris en charge et non pris en charge par la fonction dataframe_strategy lorsqu’un fichier JSON est transmis en tant qu’argument schema.

La table suivante présente les types de données Pandera pris en charge par la fonction dataframe_strategy lorsqu’elle transmet un objet DataFrameSchema en tant qu’argument schema et les types de données Snowpark auxquels ils sont mappés.

Exemples¶

Le processus typique d’utilisation de la bibliothèque Snowpark pour générer les DataFrames est le suivant :

1. Créez une fonction de test Python standard avec les différentes assertions ou conditions que votre code doit satisfaire pour toutes les entrées.

2. Ajoutez le décorateur Hypothesis @given à votre fonction de test et passez la fonction dataframe_strategy en argument. Pour plus d’informations sur le décorateur @given, voir hypothesis.given.

3. Exécutez la fonction de test. Lors de l’exécution du test, Hypothesis fournira automatiquement les entrées générées comme arguments au test.

Exemple 1 : générer les DataFrames de Snowpark à partir d’un fichier JSON

Vous trouverez ci-dessous un exemple de génération de DataFrames Snowpark à partir d’un fichier de schéma JSON généré par la fonction collect_dataframe_checkpoint du paquet 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
    ...

Exemple 2 : générer un DataFrame Snowpark à partir d’un objet DataFrameSchema Pandera

Vous trouverez ci-dessous un exemple de la manière de générer les DataFrames Snowpark à partir d’une instance de DataFrameSchema Pandera. Pour plus d’informations, voir DataFrameSchema Pandera.

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
    ...

Exemple 3 : personnaliser le comportement d’Hypothesis

Vous pouvez également personnaliser le comportement de votre test à l’aide du décorateur Hypothesis @settings. Ce décorateur vous permet de personnaliser divers paramètres de configuration afin d’adapter le comportement du test à vos besoins. En utilisant le décorateur @settings, vous pouvez contrôler des aspects tels que le nombre maximum de cas de test, la date limite pour chaque exécution de test, les niveaux de verbosité et bien d’autres. Pour plus d’informations, voir Paramètres d’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
    ...