Tests unitaires pour les notebooks

Article
08/28/2024

Vous pouvez utiliser des tests unitaires pour améliorer la qualité et la cohérence du code de vos notebooks. Le test unitaire est une approche pour tester des unités de code autonomes, telles que des fonctions, tôt dans le processus et souvent. Cela vous aide à trouver plus rapidement les problèmes sur votre code, à révéler plus tôt les prévisions erronées et à simplifier vos efforts de codage.

Cet article présente les tests unitaires de base avec fonctions. Les concepts avancés tels que les classes et interfaces de test unitaires, ainsi que l’utilisation de stubs, de mocks et de harnais de test, avec prise en charge pendant les tests unitaires pour notebooks, ne sont pas traités dans cet article. Cet article ne couvre pas non plus les autres types de méthodes de test, telles que les tests d’intégration, lestests système, les tests d’acceptation ou les méthodes de test non fonctionnelles telles que les tests de performances ou lestests de facilité d’utilisation.

Cet article présente ce qui suit :

Comment organiser les fonctions et leurs tests unitaires.
Comment écrire les fonctions dans Python, R, Scala, ainsi que les fonctions SQL définies par l’utilisateur, conçues pour être testées unitairement.
Comment appeler ces fonctions à partir de notebooks Python, R, Scala et SQL.
Comment écrire des tests unitaires dans Python, R et Scala à l’aide des frameworks de test populaires pytest pour Python, testthat pour R et ScalaTest pour Scala. En outre, comment écrire un SQL qui teste unitairement les fonctions définies par l’utilisateur SQL (UDF SQL).
Comment exécuter ces tests unitaires à partir de notebooks Python, R, Scala et SQL.

Organiser les fonctions et les tests unitaires

Il existe des méthodes communes pour organiser vos fonctions et leurs tests unitaires avec des notebooks. Chaque approche comporte ses avantages et ses inconvénients.

Pour les notebooks Python, R et Scala, les méthodes communes sont les suivantes :

Stockez les fonctions et leurs tests unitaires en dehors des notebooks..
- Avantages : vous pouvez appeler ces fonctions avec et en dehors des notebooks. Les frameworks de test sont mieux conçus pour exécuter des tests à l’extérieur des notebooks.
- Inconvénients : cette approche n’est pas prise en charge pour les notebooks Scala. Cette approche augmente également le nombre de fichiers à suivre et à gérer.
Stockez les fonctions dans un notebook et leurs tests unitaires dans un notebook distinct..
- Avantages : ces fonctions sont plus faciles à réutiliser sur les notebooks.
- Inconvénient : le nombre de notebooks à suivre et la gestion des augmentations. Ces fonctions ne peuvent pas être utilisées à l’extérieur des notebooks. Ces fonctions peuvent également être plus difficiles à tester à l’extérieur des notebooks.
Stockez les fonctions et leurs tests unitaires dans le même notebook..
- Avantages : les fonctions et leurs tests unitaires sont stockés dans un seul notebook pour faciliter le suivi et la maintenance.
- Inconvénient : ces fonctions peuvent être plus difficiles à réutiliser dans les blocs-notes. Ces fonctions ne peuvent pas être utilisées à l’extérieur des notebooks. Ces fonctions peuvent également être plus difficiles à tester à l’extérieur des notebooks.

Pour les notebooks Python et R, Databricks recommande le stockage des fonctions et de leurs tests unitaires à l’extérieur des notebooks. Pour les notebooks Scala, Databricks recommande d’inclure les fonctions dans un notebook et leurs tests unitaires dans un autre notebook distinct.

Pour les notebooks SQL, Databricks vous recommande de stocker les fonctions comme des fonctions définies par l’utilisateur SQL (UDF SQL) dans vos schémas (également appelés bases de données). Vous pouvez ensuite appeler ces UDF SQL et leurs tests unitaires à partir de notebooks SQL.

Fonctions d’écriture

Cette section décrit un ensemble d’exemples de fonctions simples qui détermine les éléments suivants :

Si un tableau existe ou non dans un tableau.
Si une colonne existe dans un tableau.
Combien de lignes existent dans une colonne pour une valeur de cette colonne.

Ces fonctions sont destinées à être simples, afin que vous puissiez vous concentrer sur les détails de tests unitaires dans cet article plutôt que les fonctions elles-mêmes.

Pour obtenir les meilleurs résultats de test unitaire, une fonction doit renvoyer un résultat prévisible unique et être d’un seul type de données. Par exemple, pour vérifier si quelque chose existe, la fonction doit renvoyer une valeur booléenne vraie ou fausse. Pour renvoyer le nombre de lignes qui existent, la fonction doit renvoyer un nombre entier non négatif. Elle ne doit pas, dans le premier exemple, renvoyer soit « faux » si quelque chose n’existe pas, soit la chose elle-même si elle n’existe pas. De même, pour le deuxième exemple, elle ne doit pas renvoyer le nombre de lignes qui existent ou « faux » si aucune ligne n’existe.

Vous pouvez ajouter ces fonctions à un espace de travail Azure Databricks existant comme suit, dans Python, R, Scala ou SQL.

Python

Le code suivant suppose que vous ayez Configuré les dossiers Databricks (référentiels), ajouté un référentiel et que le référentiel est ouvert dans votre espace de travail Azure Databricks.

Créez un fichier nommé myfunctions.py dans le référentiel et ajoutez le contenu suivant au fichier. D’autres exemples de cet article exigent que ce fichier soit nommé myfunctions.py. Vous pouvez utiliser différents noms pour vos propres fichiers.

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Because this file is not a Databricks notebook, you
# must create a Spark session. Databricks notebooks
# create a Spark session for you by default.
spark = SparkSession.builder \
                    .appName('integrity-tests') \
                    .getOrCreate()

# Does the specified table exist in the specified database?
def tableExists(tableName, dbName):
  return spark.catalog.tableExists(f"{dbName}.{tableName}")

# Does the specified column exist in the given DataFrame?
def columnExists(dataFrame, columnName):
  if columnName in dataFrame.columns:
    return True
  else:
    return False

# How many rows are there for the specified value in the specified column
# in the given DataFrame?
def numRowsInColumnForValue(dataFrame, columnName, columnValue):
  df = dataFrame.filter(col(columnName) == columnValue)

  return df.count()

R

Le code suivant suppose que vous ayez Configuré les dossiers Databricks (référentiels), ajouté un référentiel et que le référentiel est ouvert dans votre espace de travail Azure Databricks.

Créez un fichier nommé myfunctions.r dans le référentiel et ajoutez le contenu suivant au fichier. D’autres exemples de cet article exigent que ce fichier soit nommé myfunctions.r. Vous pouvez utiliser différents noms pour vos propres fichiers.

library(SparkR)

# Does the specified table exist in the specified database?
table_exists <- function(table_name, db_name) {
  tableExists(paste(db_name, ".", table_name, sep = ""))
}

# Does the specified column exist in the given DataFrame?
column_exists <- function(dataframe, column_name) {
  column_name %in% colnames(dataframe)
}

# How many rows are there for the specified value in the specified column
# in the given DataFrame?
num_rows_in_column_for_value <- function(dataframe, column_name, column_value) {
  df = filter(dataframe, dataframe[[column_name]] == column_value)

  count(df)
}

Scala

Créez un nommé notebook Scala nommé myfunctions avec le contenu suivant : D’autres exemples dans cet article exigent que ce fichier soit nommé myfunctions. Vous pouvez utiliser différents noms pour vos propres fichiers.

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.col

// Does the specified table exist in the specified database?
def tableExists(tableName: String, dbName: String) : Boolean = {
  return spark.catalog.tableExists(dbName + "." + tableName)
}

// Does the specified column exist in the given DataFrame?
def columnExists(dataFrame: DataFrame, columnName: String) : Boolean = {
  val nameOfColumn = null

  for(nameOfColumn <- dataFrame.columns) {
    if (nameOfColumn == columnName) {
      return true
    }
  }

  return false
}

// How many rows are there for the specified value in the specified column
// in the given DataFrame?
def numRowsInColumnForValue(dataFrame: DataFrame, columnName: String, columnValue: String) : Long = {
  val df = dataFrame.filter(col(columnName) === columnValue)

  return df.count()
}

SQL

Le code suivant suppose que l’échantillon de jeu de données tiers diamants dans un schéma nommé soit nommé default dans un catalogue nommé main, accessible à partir de votre espace de travail Azure Databricks. Si le catalogue ou le schéma que vous souhaitez utiliser a un nom différent, modifiez une ou les deux instructions suivantes USE pour qu’elles correspondent.

Créez un notebook SQL et ajoutez le contenu suivant à ce nouveau notebook. Attachez ensuite le notebook à un cluster et exécutez le notebook pour ajouter les UDF SQL suivants au catalogue et au schéma spécifiés.

Notes

Les UDF SQL table_exists et column_exists fonctionnent uniquement avec Unity Catalog. La prise en charge des fonctions UDF SQL pour Unity Catalog est en préversion publique.

USE CATALOG main;
USE SCHEMA default;

CREATE OR REPLACE FUNCTION table_exists(catalog_name STRING,
                                        db_name      STRING,
                                        table_name   STRING)
  RETURNS BOOLEAN
  RETURN if(
    (SELECT count(*) FROM system.information_schema.tables
     WHERE table_catalog = table_exists.catalog_name
       AND table_schema  = table_exists.db_name
       AND table_name    = table_exists.table_name) > 0,
    true,
    false
  );

CREATE OR REPLACE FUNCTION column_exists(catalog_name STRING,
                                         db_name      STRING,
                                         table_name   STRING,
                                         column_name  STRING)
  RETURNS BOOLEAN
  RETURN if(
    (SELECT count(*) FROM system.information_schema.columns
     WHERE table_catalog = column_exists.catalog_name
       AND table_schema  = column_exists.db_name
       AND table_name    = column_exists.table_name
       AND column_name   = column_exists.column_name) > 0,
    true,
    false
  );

CREATE OR REPLACE FUNCTION num_rows_for_clarity_in_diamonds(clarity_value STRING)
  RETURNS BIGINT
  RETURN SELECT count(*)
         FROM main.default.diamonds
         WHERE clarity = clarity_value

Fonctions d’appel

Cette section présente le code d’appel des fonctions précédentes. Vous pouvez utiliser ces fonctions, par exemple, pour compter le nombre de lignes dans un tableau où une valeur spécifiée existe dans une colonne spécifiée. Toutefois, vous devriez vérifier si le tableau et la colonne existent réellement dans ce tableau avant de continuer. Le code suivant vérifie ces conditions.

Si vous avez ajouté les fonctions de la section précédente à votre espace de travail Azure Databricks, vous pouvez appeler ces fonctions à partir de votre espace de travail comme suit.

Python

Créez un notebook Python dans le même dossier que le fichier myfunctions.pyprécédent de votre référentiel, et ajoutez le contenu suivant au notebook. Modifiez les valeurs de variable pour le nom de la table, le nom du schéma (base de données), le nom de la colonne et la valeur de colonne si nécessaire. Ensuite, attachez le notebook à un cluster et exécutez le notebook pour afficher les résultats.

from myfunctions import *

tableName   = "diamonds"
dbName      = "default"
columnName  = "clarity"
columnValue = "VVS2"

# If the table exists in the specified database...
if tableExists(tableName, dbName):

  df = spark.sql(f"SELECT * FROM {dbName}.{tableName}")

  # And the specified column exists in that table...
  if columnExists(df, columnName):
    # Then report the number of rows for the specified value in that column.
    numRows = numRowsInColumnForValue(df, columnName, columnValue)

    print(f"There are {numRows} rows in '{tableName}' where '{columnName}' equals '{columnValue}'.")
  else:
    print(f"Column '{columnName}' does not exist in table '{tableName}' in schema (database) '{dbName}'.")
else:
  print(f"Table '{tableName}' does not exist in schema (database) '{dbName}'.")

R

Créez un notebook R dans le même dossier que le fichier myfunctions.r précédent de votre référentiel, et ajoutez le contenu suivant au notebook. Modifiez les valeurs de variable pour le nom de la table, le nom du schéma (base de données), le nom de la colonne et la valeur de colonne si nécessaire. Ensuite, attachez le notebook à un cluster et exécutez le notebook pour afficher les résultats.

library(SparkR)
source("myfunctions.r")

table_name   <- "diamonds"
db_name      <- "default"
column_name  <- "clarity"
column_value <- "VVS2"

# If the table exists in the specified database...
if (table_exists(table_name, db_name)) {

  df = sql(paste("SELECT * FROM ", db_name, ".", table_name, sep = ""))

  # And the specified column exists in that table...
  if (column_exists(df, column_name)) {
    # Then report the number of rows for the specified value in that column.
    num_rows = num_rows_in_column_for_value(df, column_name, column_value)

    print(paste("There are ", num_rows, " rows in table '", table_name, "' where '", column_name, "' equals '", column_value, "'.", sep = "")) 
  } else {
    print(paste("Column '", column_name, "' does not exist in table '", table_name, "' in schema (database) '", db_name, "'.", sep = ""))
  }

} else {
  print(paste("Table '", table_name, "' does not exist in schema (database) '", db_name, "'.", sep = ""))
}

Scala

Créez un autre notebook Scala dans le même dossier que le notebook Scala précédent myfunctions et ajoutez le contenu suivant à ce nouveau notebook.

Dans la première cellule de ce nouveau notebook, ajoutez le code suivant, qui appelle %run magic. Ce magic rend disponible le contenu du notebook myfunctions pour votre nouveau notebook.

%run ./myfunctions

Dans la deuxième cellule de ce nouveau notebook, ajoutez le code suivant. Modifiez les valeurs de variable pour le nom de la table, le nom du schéma (base de données), le nom de la colonne et la valeur de colonne si nécessaire. Ensuite, attachez le notebook à un cluster et exécutez le notebook pour afficher les résultats.

val tableName   = "diamonds"
val dbName      = "default"
val columnName  = "clarity"
val columnValue = "VVS2"

// If the table exists in the specified database...
if (tableExists(tableName, dbName)) {

  val df = spark.sql("SELECT * FROM " + dbName + "." + tableName)

  // And the specified column exists in that table...
  if (columnExists(df, columnName)) {
    // Then report the number of rows for the specified value in that column.
    val numRows = numRowsInColumnForValue(df, columnName, columnValue)

    println("There are " + numRows + " rows in '" + tableName + "' where '" + columnName + "' equals '" + columnValue + "'.")
  } else {
    println("Column '" + columnName + "' does not exist in table '" + tableName + "' in database '" + dbName + "'.")
  }

} else {
  println("Table '" + tableName + "' does not exist in database '" + dbName + "'.")
}

SQL

Ajoutez le code suivant à une nouvelle cellule du notebook précédent ou à une autre cellule dans un notebook distinct. Modifiez les noms du schéma ou du catalogue si nécessaire pour qu’ils correspondent aux vôtres, puis exécutez cette cellule pour afficher les résultats.

SELECT CASE
-- If the table exists in the specified catalog and schema...
WHEN
  table_exists("main", "default", "diamonds")
THEN
  -- And the specified column exists in that table...
  (SELECT CASE
   WHEN
     column_exists("main", "default", "diamonds", "clarity")
   THEN
     -- Then report the number of rows for the specified value in that column.
     printf("There are %d rows in table 'main.default.diamonds' where 'clarity' equals 'VVS2'.",
            num_rows_for_clarity_in_diamonds("VVS2"))
   ELSE
     printf("Column 'clarity' does not exist in table 'main.default.diamonds'.")
   END)
ELSE
  printf("Table 'main.default.diamonds' does not exist.")
END

Écrire des tests unitaires

Cette section présente les codes qui testent chacune des fonctions décrites au début de cet article. Si plus tard, vous apportez des modifications aux fonctions, vous pouvez utiliser des tests unitaires pour déterminer si ces fonctions fonctionnent toujours comme prévu.

Si vous avez ajouté les fonctions présentées au début de cet article à votre espace de travail Azure Databricks, vous pouvez y ajouter les tests unitaires de ces fonctions comme suit.

Python

Créez un autre fichier nommé test_myfunctions.py dans le même dossier que le fichier myfunctions.py précédent dans votre référentiel, et ajoutez les contenus suivants au notebook. Par défaut, pytest recherche les fichiers .py à tester dont les noms commencent par test_ (ou se terminent par _test). De même, par défaut, pytest recherche à l’intérieur de ces fichiers les fonctions à tester dont les noms commencent par test_.

En général, il est recommandé de ne pas exécuter de tests unitaires sur des fonctions qui fonctionnent avec des données en production. Cela est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient des données. Pour protéger vos données de production contre toute compromission par vos tests unitaires de manière inattendue, vous devez exécuter des tests unitaires sur des données hors production. Une approche courante consiste à créer des données factices aussi proches que possible des données de production. L’exemple de code suivant crée de fausses données pour les tests unitaires à exécuter.

import pytest
import pyspark
from myfunctions import *
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType

tableName    = "diamonds"
dbName       = "default"
columnName   = "clarity"
columnValue  = "SI2"

# Because this file is not a Databricks notebook, you
# must create a Spark session. Databricks notebooks
# create a Spark session for you by default.
spark = SparkSession.builder \
                    .appName('integrity-tests') \
                    .getOrCreate()

# Create fake data for the unit tests to run against.
# In general, it is a best practice to not run unit tests
# against functions that work with data in production.
schema = StructType([ \
  StructField("_c0",     IntegerType(), True), \
  StructField("carat",   FloatType(),   True), \
  StructField("cut",     StringType(),  True), \
  StructField("color",   StringType(),  True), \
  StructField("clarity", StringType(),  True), \
  StructField("depth",   FloatType(),   True), \
  StructField("table",   IntegerType(), True), \
  StructField("price",   IntegerType(), True), \
  StructField("x",       FloatType(),   True), \
  StructField("y",       FloatType(),   True), \
  StructField("z",       FloatType(),   True), \
])

data = [ (1, 0.23, "Ideal",   "E", "SI2", 61.5, 55, 326, 3.95, 3.98, 2.43 ), \
         (2, 0.21, "Premium", "E", "SI1", 59.8, 61, 326, 3.89, 3.84, 2.31 ) ]

df = spark.createDataFrame(data, schema)

# Does the table exist?
def test_tableExists():
  assert tableExists(tableName, dbName) is True

# Does the column exist?
def test_columnExists():
  assert columnExists(df, columnName) is True

# Is there at least one row for the value in the specified column?
def test_numRowsInColumnForValue():
  assert numRowsInColumnForValue(df, columnName, columnValue) > 0

R

Créez un autre fichier nommé test_myfunctions.r dans le même dossier que le fichier myfunctions.r précédent dans votre référentiel, et ajoutez les contenus suivants au notebook. Par défaut, testthat recherche les fichiers .r à tester dont les noms commencent par test.

library(testthat)
source("myfunctions.r")

table_name   <- "diamonds"
db_name      <- "default"
column_name  <- "clarity"
column_value <- "SI2"

# Create fake data for the unit tests to run against.
# In general, it is a best practice to not run unit tests
# against functions that work with data in production.
schema <- structType(
  structField("_c0",     "integer"),
  structField("carat",   "float"),
  structField("cut",     "string"),
  structField("color",   "string"),
  structField("clarity", "string"),
  structField("depth",   "float"),
  structField("table",   "integer"),
  structField("price",   "integer"),
  structField("x",       "float"),
  structField("y",       "float"),
  structField("z",       "float"))

data <- list(list(as.integer(1), 0.23, "Ideal",   "E", "SI2", 61.5, as.integer(55), as.integer(326), 3.95, 3.98, 2.43),
             list(as.integer(2), 0.21, "Premium", "E", "SI1", 59.8, as.integer(61), as.integer(326), 3.89, 3.84, 2.31))

df <- createDataFrame(data, schema)

# Does the table exist?
test_that ("The table exists.", {
  expect_true(table_exists(table_name, db_name))
})

# Does the column exist?
test_that ("The column exists in the table.", {
  expect_true(column_exists(df, column_name))
})

# Is there at least one row for the value in the specified column?
test_that ("There is at least one row in the query result.", {
  expect_true(num_rows_in_column_for_value(df, column_name, column_value) > 0)
})

Scala

Créez un autre notebook Scala dans le même dossier que le notebook Scala précédent myfunctions et ajoutez le contenu suivant à ce nouveau notebook.

Dans la première cellule de ce nouveau notebook, ajoutez le code suivant, qui appelle %run magic. Ce magic rend disponible le contenu du notebook myfunctions pour votre nouveau notebook.

%run ./myfunctions

Dans la deuxième cellule, ajoutez le code suivant. Ce code définit vos tests unitaires et spécifie comment les exécuter.

import org.scalatest._
import org.apache.spark.sql.types.{StructType, StructField, IntegerType, FloatType, StringType}
import scala.collection.JavaConverters._

class DataTests extends AsyncFunSuite {

  val tableName   = "diamonds"
  val dbName      = "default"
  val columnName  = "clarity"
  val columnValue = "SI2"

  // Create fake data for the unit tests to run against.
  // In general, it is a best practice to not run unit tests
  // against functions that work with data in production.
  val schema = StructType(Array(
                 StructField("_c0",     IntegerType),
                 StructField("carat",   FloatType),
                 StructField("cut",     StringType),
                 StructField("color",   StringType),
                 StructField("clarity", StringType),
                 StructField("depth",   FloatType),
                 StructField("table",   IntegerType),
                 StructField("price",   IntegerType),
                 StructField("x",       FloatType),
                 StructField("y",       FloatType),
                 StructField("z",       FloatType)
               ))

  val data = Seq(
                  Row(1, 0.23, "Ideal",   "E", "SI2", 61.5, 55, 326, 3.95, 3.98, 2.43),
                  Row(2, 0.21, "Premium", "E", "SI1", 59.8, 61, 326, 3.89, 3.84, 2.31)
                ).asJava

  val df = spark.createDataFrame(data, schema)

  // Does the table exist?
  test("The table exists") {
    assert(tableExists(tableName, dbName) == true)
  }

  // Does the column exist?
  test("The column exists") {
    assert(columnExists(df, columnName) == true)
  }

  // Is there at least one row for the value in the specified column?
  test("There is at least one matching row") {
    assert(numRowsInColumnForValue(df, columnName, columnValue) > 0)
  }
}

nocolor.nodurations.nostacks.stats.run(new DataTests)

Notes

Cet exemple de code utilise le style de test FunSuite dans ScalaTest. Pour d’autres styles de test disponibles, consultez Sélection des styles de test pour votre projet.

SQL

Avant d’ajouter des tests unitaires, vous devez savoir qu’en général, il est recommandé de ne pas exécuter de tests unitaires sur des fonctions qui fonctionnent avec des données en production. Cela est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient des données. Pour protéger vos données de production contre toute compromission par vos tests unitaires de manière inattendue, vous devez exécuter des tests unitaires sur des données hors production. Une approche courante consiste à exécuter des tests unitaires sur des vues plutôt que sur des tables.

Pour créer une vue, vous pouvez appeler la commande CREATE VIEW à partir d’une nouvelle cellule du bloc-notes précédent ou d’un bloc-notes distinct. L’exemple suivant suppose que vous disposez d’une table existante nommée diamonds dans un schéma (base de données) nommé default dans un catalogue nommé main. Modifiez ces noms pour qu’ils correspondent aux vôtres en fonction des besoins, puis exécutez uniquement cette cellule.

USE CATALOG main;
USE SCHEMA default;

CREATE VIEW view_diamonds AS
SELECT * FROM diamonds;

Après avoir créé la vue, ajoutez chacune des déclarations suivantesSELECT à sa propre nouvelle cellule dans le cahier précédent ou à sa propre nouvelle cellule dans un cahier séparé. Modifiez les noms pour qu’ils correspondent aux vôtres en fonction des besoins.

SELECT if(table_exists("main", "default", "view_diamonds"),
          printf("PASS: The table 'main.default.view_diamonds' exists."),
          printf("FAIL: The table 'main.default.view_diamonds' does not exist."));

SELECT if(column_exists("main", "default", "view_diamonds", "clarity"),
          printf("PASS: The column 'clarity' exists in the table 'main.default.view_diamonds'."),
          printf("FAIL: The column 'clarity' does not exists in the table 'main.default.view_diamonds'."));

SELECT if(num_rows_for_clarity_in_diamonds("VVS2") > 0,
          printf("PASS: The table 'main.default.view_diamonds' has at least one row where the column 'clarity' equals 'VVS2'."),
          printf("FAIL: The table 'main.default.view_diamonds' does not have at least one row where the column 'clarity' equals 'VVS2'."));

Exécuter des tests unitaires

Cette section explique comment exécuter les tests unitaires que vous avez codés dans la section précédente. Lorsque vous exécutez les tests unitaires, vous obtenez des résultats montrant quels tests unitaires ont réussi et lesquels ont échoué.

Si vous avez ajouté les fonctions de la section précédente à votre espace de travail Azure Databricks, vous pouvez exécuter ces tests unitaires à partir de votre espace de travail. Vous pouvez exécuter ces tests unitaires soit manuellement ou selon une planification.

Python

Créez un notebook Python dans le même dossier que le fichier test_myfunctions.py précédent de votre référentiel, et ajoutez le contenu suivant au notebook.

Dans la première cellule du notebook, ajoutez le code suivant, puis exécutez la cellule, qui appelle le %pip magic. Cette magie installe pytest.

%pip install pytest

Dans la deuxième cellule, ajoutez le code suivant, puis exécutez la cellule. Les résultats montrent quels tests unitaires ont réussi et lesquels ont échoué.

import pytest
import sys

# Skip writing pyc files on a readonly filesystem.
sys.dont_write_bytecode = True

# Run pytest.
retcode = pytest.main([".", "-v", "-p", "no:cacheprovider"])

# Fail the cell execution if there are any test failures.
assert retcode == 0, "The pytest invocation failed. See the log for details."

R

Créez un notebook R dans le même dossier que le fichier test_myfunctions.r précédent de votre référentiel, et ajoutez le contenu suivant.

Dans la première cellule, ajoutez le code suivant, puis exécutez la cellule, qui appelle la fonctioninstall.packages. La fonction installe testthat.

install.packages("testthat")

Dans la deuxième cellule, ajoutez le code suivant, puis exécutez la cellule. Les résultats montrent quels tests unitaires ont réussi et lesquels ont échoué.

library(testthat)
source("myfunctions.r")

test_dir(".", reporter = "tap")

Scala

Exécutez la première puis la deuxième cellule du notebook de la section précédente. Les résultats montrent quels tests unitaires ont réussi et lesquels ont échoué.

SQL

Exécutez chacune des trois cellules du notebook de la section précédente. Les résultats indiquent pour chaque test unitaire s’il a réussi ou échoué.

Si vous n’avez plus besoin de la vue après avoir exécuté vos tests unitaires, vous pouvez supprimer l’affichage. Pour supprimer cette vue, vous pouvez ajouter le code suivant à une nouvelle cellule dans l’un des blocs-notes précédents, puis exécuter uniquement cette cellule.

DROP VIEW view_diamonds;

Conseil

Vous pouvez afficher les résultats de vos exécutions de notebook (y compris les résultats des tests unitaires) dans les journaux des pilotes de votre cluster. Vous pouvez également spécifier un emplacement pour la remise des journaux de votre cluster.

Vous pouvez configurer un système d’intégration continue et de livraison ou de déploiement continu (CI/CD), tel que GitHub Actions, pour exécuter automatiquement vos tests unitaires chaque fois que votre code change. Pour obtenir un exemple, consultez la couverture des GitHub Actions dans Meilleures pratiques d’ingénierie logicielle pour les notebooks.

Partager via

Tests unitaires pour les notebooks

Organiser les fonctions et les tests unitaires

Fonctions d’écriture

Python

R

Scala

SQL

Fonctions d’appel

Python

R

Scala

SQL

Écrire des tests unitaires

Python

R

Scala

SQL

Exécuter des tests unitaires

Python

R

Scala

SQL

Ressources supplémentaires

pytest

testthat

Scalatest

SQL

Commentaires

Ressources supplémentaires