Unit test per i notebook

Articolo
08/28/2024

È possibile usare unit test per migliorare la qualità e la coerenza del codice dei notebook. Gli unit test sono un approccio per testare unità di codice autonome, ad esempio funzioni, in anticipo e spesso. In questo modo è possibile trovare più rapidamente i problemi del codice, individuare ipotesi sbagliate sul codice prima e semplificare le attività di codifica complessive.

Questo articolo è un'introduzione agli unit test di base con le funzioni. Concetti avanzati, ad esempio classi e interfacce di unit test, nonché l'uso di stub, simulazioni e test harness, oltre a supportare l'esecuzione di unit test per i notebook, non rientrano nell'ambito di questo articolo. Questo articolo non tratta anche altri tipi di metodi di test, ad esempio test di integrazione, test di sistema, test di accettazione o metodi di test non funzionali, ad esempio test delle prestazioni o test di usabilità.

Questo articolo illustra quanto segue:

Come organizzare le funzioni e i relativi unit test.
Come scrivere funzioni in Python, R, Scala e funzioni definite dall'utente in SQL, ben progettate per essere sottoposte a unit test.
Come chiamare queste funzioni da notebook Python, R, Scala e SQL.
Come scrivere unit test in Python, R e Scala usando i framework di test più diffusi pytest per Python, testatat per R e ScalaTest per Scala. Inoltre, come scrivere SQL che esegue unit test delle funzioni DEFINITE dall'utente (UDF SQL).
Come eseguire questi unit test da notebook Python, R, Scala e SQL.

Organizzare funzioni e unit test

Esistono alcuni approcci comuni per organizzare le funzioni e i relativi unit test con i notebook. Ogni approccio presenta vantaggi e sfide.

Per i notebook Python, R e Scala, gli approcci comuni includono quanto segue:

Archiviare le funzioni e i relativi unit test all'esterno dei notebook.
- Vantaggi: è possibile chiamare queste funzioni con e all'esterno dei notebook. I framework di test sono progettati meglio per eseguire test all'esterno dei notebook.
- Sfide: questo approccio non è supportato per i notebook Scala. Questo approccio aumenta anche il numero di file da tenere traccia e gestire.
Archiviare le funzioni in un notebook e i relativi unit test in un notebook separato.
- Vantaggi: queste funzioni sono più facili da riutilizzare nei notebook.
- Sfide: il numero di notebook da tenere traccia e gestire aumenta. Queste funzioni non possono essere usate all'esterno dei notebook. Queste funzioni possono anche essere più difficili da testare all'esterno dei notebook.
Archiviare le funzioni e i relativi unit test all'interno dello stesso notebook.
- Vantaggi: le funzioni e i relativi unit test vengono archiviati all'interno di un singolo notebook per semplificare il rilevamento e la manutenzione.
- Sfide: queste funzioni possono essere più difficili da riutilizzare tra notebook. Queste funzioni non possono essere usate all'esterno dei notebook. Queste funzioni possono anche essere più difficili da testare all'esterno dei notebook.

Per i notebook Python e R, Databricks consiglia di archiviare le funzioni e i relativi unit test all'esterno dei notebook. Per i notebook Scala, Databricks consiglia di includere funzioni in un unico notebook e i relativi unit test in un notebook separato.

Per i notebook SQL, Databricks consiglia di archiviare le funzioni come funzioni definite dall'utente SQL (UDF SQL) negli schemi (noti anche come database). È quindi possibile chiamare queste funzioni definite dall'utente SQL e i relativi unit test dai notebook SQL.

Scrivere funzioni

Questa sezione descrive un semplice set di funzioni di esempio che determinano quanto segue:

Indica se esiste una tabella in un database.
Indica se una colonna esiste in una tabella.
Numero di righe presenti in una colonna per un valore all'interno di tale colonna.

Queste funzioni sono concepite per essere semplici, in modo che sia possibile concentrarsi sui dettagli degli unit test in questo articolo anziché concentrarsi sulle funzioni stesse.

Per ottenere i migliori risultati degli unit test, una funzione deve restituire un singolo risultato prevedibile e essere di un singolo tipo di dati. Ad esempio, per verificare se esiste qualcosa, la funzione deve restituire un valore booleano true o false. Per restituire il numero di righe esistenti, la funzione deve restituire un numero intero non negativo. Non deve, nel primo esempio, restituire false se qualcosa non esiste o la cosa stessa se esiste. Analogamente, per il secondo esempio, non deve restituire il numero di righe esistenti o false se non esistono righe.

È possibile aggiungere queste funzioni a un'area di lavoro di Azure Databricks esistente come indicato di seguito, in Python, R, Scala o SQL.

Python

Il codice seguente presuppone che siano state configurate le cartelle Git di Databricks (Repos), sia stato aggiunto un repository e che il repository sia aperto nell'area di lavoro di Azure Databricks.

Creare un file denominato myfunctions.py all'interno del repository e aggiungere il contenuto seguente al file. Altri esempi in questo articolo prevedono che questo file sia denominato myfunctions.py. È possibile usare nomi diversi per i propri file.

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

Creare un file denominato myfunctions.r all'interno del repository e aggiungere il contenuto seguente al file. Altri esempi in questo articolo prevedono che questo file sia denominato myfunctions.r. È possibile usare nomi diversi per i propri file.

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

Creare un notebook Scala denominato myfunctions con il contenuto seguente. Altri esempi in questo articolo prevedono che questo notebook sia denominato myfunctions. È possibile usare nomi diversi per i propri notebook.

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

Il codice seguente presuppone che siano presenti diamanti del set di dati di esempio di terze parti all'interno di uno schema denominato default all'interno di un catalogo denominato main accessibile dall'area di lavoro di Azure Databricks. Se il catalogo o lo schema da usare ha un nome diverso, modificare una o entrambe le istruzioni seguenti USE in modo che corrispondano.

Creare un notebook SQL e aggiungere il contenuto seguente a questo nuovo notebook. Collegare quindi il notebook a un cluster ed eseguire il notebook per aggiungere le funzioni definite dall'utente SQL seguenti al catalogo e allo schema specificati.

Nota

Le funzioni definite table_exists dall'utente di SQL e column_exists funzionano solo con il catalogo Unity. Il supporto di funzioni definite dall'utente di SQL per il catalogo Unity è disponibile in anteprima pubblica.

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

Chiamare le funzioni

Questa sezione descrive il codice che chiama le funzioni precedenti. È possibile usare queste funzioni, ad esempio, per contare il numero di righe nella tabella in cui esiste un valore specificato all'interno di una colonna specificata. Tuttavia, è consigliabile verificare se la tabella esiste effettivamente e se la colonna esiste effettivamente in tale tabella prima di procedere. Il codice seguente verifica la presenza di queste condizioni.

Se sono state aggiunte le funzioni dalla sezione precedente all'area di lavoro di Azure Databricks, è possibile chiamare queste funzioni dall'area di lavoro come indicato di seguito.

Python

Creare un notebook Python nella stessa cartella del file precedente myfunctions.py nel repository e aggiungere il contenuto seguente al notebook. Modificare i valori delle variabili per il nome della tabella, il nome dello schema (database), il nome della colonna e il valore della colonna in base alle esigenze. Collegare quindi il notebook a un cluster ed eseguire il notebook per visualizzare i risultati.

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

Creare un notebook R nella stessa cartella del file precedente myfunctions.r nel repository e aggiungere il contenuto seguente al notebook. Modificare i valori delle variabili per il nome della tabella, il nome dello schema (database), il nome della colonna e il valore della colonna in base alle esigenze. Collegare quindi il notebook a un cluster ed eseguire il notebook per visualizzare i risultati.

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

Creare un altro notebook Scala nella stessa cartella del notebook Scala precedente myfunctions e aggiungere il contenuto seguente a questo nuovo notebook.

Nella prima cella del nuovo notebook aggiungere il codice seguente, che chiama il magic %run . Questa magia rende disponibile il contenuto del notebook per il myfunctions nuovo notebook.

%run ./myfunctions

Nella seconda cella del nuovo notebook aggiungere il codice seguente. Modificare i valori delle variabili per il nome della tabella, il nome dello schema (database), il nome della colonna e il valore della colonna in base alle esigenze. Collegare quindi il notebook a un cluster ed eseguire il notebook per visualizzare i risultati.

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

Aggiungere il codice seguente a una nuova cella del notebook precedente o a una cella in un notebook separato. Modificare lo schema o i nomi di catalogo, se necessario, per trovare le corrispondenze con l'utente e quindi eseguire questa cella per visualizzare i risultati.

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

Scrivere unit test

In questa sezione viene descritto il codice che testa ognuna delle funzioni descritte all'inizio di questo articolo. Se si apportano modifiche alle funzioni in futuro, è possibile usare unit test per determinare se tali funzioni funzionano ancora come previsto.

Se all'inizio di questo articolo sono state aggiunte le funzioni all'area di lavoro di Azure Databricks, è possibile aggiungere unit test per queste funzioni all'area di lavoro come indicato di seguito.

Python

Creare un altro file denominato test_myfunctions.py nella stessa cartella del file precedente myfunctions.py nel repository e aggiungere il contenuto seguente al file. Per impostazione predefinita, pytest cerca i .py file i cui nomi iniziano con test_ (o terminano con _test) da testare. Analogamente, per impostazione predefinita, pytest cerca all'interno di questi file funzioni i cui nomi iniziano con test_ per eseguire il test.

In generale, è consigliabile non eseguire unit test su funzioni che funzionano con i dati nell'ambiente di produzione. Ciò è particolarmente importante per le funzioni che aggiungono, rimuovono o modificano i dati. Per proteggere i dati di produzione da essere compromessi dagli unit test in modi imprevisti, è consigliabile eseguire unit test su dati non di produzione. Un approccio comune consiste nel creare dati falsi il più vicino possibile ai dati di produzione. Nell'esempio di codice seguente vengono creati dati falsi per l'esecuzione degli unit test.

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

Creare un altro file denominato test_myfunctions.r nella stessa cartella del file precedente myfunctions.r nel repository e aggiungere il contenuto seguente al file. Per impostazione predefinita, testthat cerca i .r file i cui nomi iniziano con test per eseguire il 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

Creare un altro notebook Scala nella stessa cartella del notebook Scala precedente myfunctions e aggiungere il contenuto seguente a questo nuovo notebook.

Nella prima cella del nuovo notebook aggiungere il codice seguente, che chiama il %run magic. Questa magia rende disponibile il contenuto del notebook per il myfunctions nuovo notebook.

%run ./myfunctions

Nella seconda cella aggiungere il codice seguente. Questo codice definisce gli unit test e specifica come eseguirli.

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)

Nota

Questo esempio di codice usa lo FunSuite stile di test in ScalaTest. Per altri stili di test disponibili, vedere Selezione degli stili di test per il progetto.

SQL

Prima di aggiungere unit test, è consigliabile tenere presente che, in generale, è consigliabile non eseguire unit test sulle funzioni che funzionano con i dati nell'ambiente di produzione. Ciò è particolarmente importante per le funzioni che aggiungono, rimuovono o modificano i dati. Per proteggere i dati di produzione da essere compromessi dagli unit test in modi imprevisti, è consigliabile eseguire unit test su dati non di produzione. Un approccio comune consiste nell'eseguire unit test sulle viste anziché sulle tabelle.

Per creare una vista, è possibile chiamare il comando CREATE VIEW da una nuova cella nel notebook precedente o in un notebook separato. Nell'esempio seguente si presuppone che sia presente una tabella esistente denominata diamonds all'interno di uno schema (database) denominato default all'interno di un catalogo denominato main. Modificare questi nomi in modo che corrispondano al proprio in base alle esigenze e quindi eseguire solo tale cella.

USE CATALOG main;
USE SCHEMA default;

CREATE VIEW view_diamonds AS
SELECT * FROM diamonds;

Dopo aver creato la vista, aggiungere ognuna delle istruzioni seguenti SELECT alla propria nuova cella nel notebook precedente o alla propria nuova cella in un notebook separato. Modificare i nomi in modo che corrispondano ai propri in base alle esigenze.

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'."));

Esecuzione di unit test

Questa sezione descrive come eseguire gli unit test codificati nella sezione precedente. Quando si eseguono gli unit test, si ottengono risultati che mostrano gli unit test superati e non riusciti.

Se gli unit test sono stati aggiunti dalla sezione precedente all'area di lavoro di Azure Databricks, è possibile eseguire questi unit test dall'area di lavoro. È possibile eseguire questi unit test manualmente o in base a una pianificazione.

Python

Creare un notebook Python nella stessa cartella del file precedente test_myfunctions.py nel repository e aggiungere il contenuto seguente.

Nella prima cella del nuovo notebook aggiungere il codice seguente e quindi eseguire la cella, che chiama il %pip magic. Questo magic installa pytest.

%pip install pytest

Nella seconda cella aggiungere il codice seguente e quindi eseguire la cella. I risultati mostrano gli unit test superati e non riusciti.

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

Creare un notebook R nella stessa cartella del file precedente test_myfunctions.r nel repository e aggiungere il contenuto seguente.

Nella prima cella aggiungere il codice seguente e quindi eseguire la cella , che chiama la install.packages funzione . Questa funzione installa testthat.

install.packages("testthat")

Nella seconda cella aggiungere il codice seguente e quindi eseguire la cella. I risultati mostrano gli unit test superati e non riusciti.

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

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

Scala

Eseguire la prima e quindi le seconde celle del notebook dalla sezione precedente. I risultati mostrano gli unit test superati e non riusciti.

SQL

Eseguire ognuna delle tre celle del notebook dalla sezione precedente. I risultati indicano se ogni unit test è stato superato o non è riuscito.

Se la visualizzazione non è più necessaria dopo l'esecuzione degli unit test, è possibile eliminare la visualizzazione. Per eliminare questa visualizzazione, è possibile aggiungere il codice seguente a una nuova cella all'interno di uno dei notebook precedenti e quindi eseguire solo tale cella.

DROP VIEW view_diamonds;

Suggerimento

È possibile visualizzare i risultati delle esecuzioni del notebook (inclusi i risultati degli unit test) nei log dei driver del cluster. È anche possibile specificare un percorso per il recapito dei log del cluster.

È possibile configurare un sistema di integrazione continua e distribuzione continua (CI/CD), ad esempio GitHub Actions, per eseguire automaticamente gli unit test ogni volta che il codice cambia. Per un esempio, vedere la copertura di GitHub Actions nelle procedure consigliate per la progettazione software per i notebook.

Condividi tramite

Unit test per i notebook

Organizzare funzioni e unit test

Scrivere funzioni

Python

R

Scala

SQL

Chiamare le funzioni

Python

R

Scala

SQL

Scrivere unit test

Python

R

Scala

SQL

Esecuzione di unit test

Python

R

Scala

SQL

Risorse aggiuntive

pytest

testthat

ScalaTest

SQL

Commenti e suggerimenti

Risorse aggiuntive