Prueba unitaria para cuadernos
Puede usar la prueba unitaria para ayudar a mejorar la calidad y la coherencia del código de los cuadernos. La prueba unitaria es un enfoque para probar unidades de código independientes, como las funciones, desde el primer momento y con frecuencia. Esto permite encontrar problemas con el código más rápido, descubrir suposiciones erróneas sobre el código antes y simplificar los esfuerzos generales de codificación.
Este artículo es una introducción a la prueba unitaria básica con funciones. Conceptos avanzados, como las clases e interfaces de prueba unitaria, así como el uso de códigos auxiliares, simulaciones y herramientas de ejecución de pruebas, si bien también se admiten cuando se realiza la prueba unitaria para cuadernos, están fuera del ámbito de este artículo. En este artículo tampoco se tratan otros tipos de métodos de prueba, como pruebas de integración, pruebas del sistema, pruebas de aceptación o métodos de pruebas no funcionales, como pruebas de rendimiento o pruebas de facilidad de uso.
En este artículo se muestra lo siguiente:
- Procedimientos para organizar las funciones y su prueba unitaria.
- Procedimientos para escribir funciones en Python, R, Scala, así como funciones definidas por el usuario en SQL, que están bien diseñadas para realizar la prueba unitaria.
- Procedimientos para llamar a estas funciones desde cuadernos de Python, R, Scala y SQL.
- Procedimientos para escribir pruebas unitarias en Python, R y Scala mediante los marcos de pruebas populares pytest para Python, testthat para R y ScalaTest para Scala. También cómo escribir SQL que realice pruebas unitarias de funciones definidas por el usuario de SQL (UDF de SQL).
- Procedimientos para ejecutar estas pruebas unitarias desde cuadernos de Python, R, Scala y SQL.
Organización de funciones y pruebas unitarias
Hay algunos enfoques comunes para organizar las funciones y sus pruebas unitarias con cuadernos. Cada uno de ellos presenta ventajas y desafíos.
En el caso de los cuadernos de Python, R y Scala, los enfoques comunes incluyen lo siguiente:
- Almacenar las funciones y sus pruebas unitarias fuera de los cuadernos..
- Ventajas: puede llamar a estas funciones con los cuadernos y fuera de ellos. Los marcos de pruebas están mejor diseñados para ejecutar pruebas fuera de cuadernos.
- Desafíos: este enfoque no es compatible con cuadernos de Scala. Este enfoque también aumenta el número de archivos en los que se tiene que realizar un seguimiento y mantenimiento.
- Almacenar funciones en un cuaderno y sus pruebas unitarias en otro cuaderno independiente..
- Ventajas: estas funciones son más fáciles de reutilizar entre cuadernos.
- Desafíos: aumenta el número de cuadernos en los que se tiene que realizar un seguimiento y mantenimiento. Estas funciones no se pueden usar fuera de los cuadernos. Estas funciones también pueden ser más difíciles de probar fuera de los cuadernos.
- Almacenar las funciones y sus pruebas unitarias en el mismo cuaderno..
- Ventajas: las funciones y sus pruebas unitarias se almacenan en un único cuaderno para facilitar el seguimiento y el mantenimiento.
- Desafíos: estas funciones pueden ser más difíciles de reutilizar entre cuadernos. Estas funciones no se pueden usar fuera de los cuadernos. Estas funciones también pueden ser más difíciles de probar fuera de los cuadernos.
En el caso de los cuadernos de Python y R, Databricks recomienda almacenar funciones y sus pruebas unitarias fuera de los cuadernos. En el caso de los cuadernos de Scala, Databricks recomienda incluir funciones en un cuaderno y sus pruebas unitarias en otro cuaderno independiente.
En el caso de los cuadernos de SQL, Databricks recomienda almacenar funciones que sean definidas por el usuario de SQL (UDF de SQL) en los esquemas (también conocidos como bases de datos). Después, puede llamar a estas UDF de SQL y sus pruebas unitarias desde cuadernos de SQL.
Funciones de escritura
En esta sección se describe un conjunto sencillo de funciones de ejemplo que determinan lo siguiente:
- Si existe una tabla en una base de datos.
- Si existe una columna en una tabla.
- Cuántas filas existen en una columna para un valor en esa columna.
Estas funciones están diseñadas para ser sencillas, de modo que pueda centrarse en los detalles de la prueba unitaria en este artículo en lugar de centrarse en las propias funciones.
Para obtener los mejores resultados de la prueba unitaria, una función debe devolver un único resultado predecible y este debe ser de un único tipo de datos. Por ejemplo, para comprobar si existe algo, la función debe devolver un valor booleano de true o false. Para devolver el número de filas que existen, la función debe devolver un número entero no negativo. En el primer ejemplo, no debería devolver false si algo no existe o lo mismo en caso contrario. Del mismo modo, en el segundo ejemplo, no debe devolver el número de filas que existen o false si no existen filas.
Puede agregar estas funciones a un área de trabajo de Azure Databricks existente, tal como se indica a continuación en Phyton, R, Scala o SQL.
Python
En el código siguiente se supone que ha configurado carpetas de Git de Databricks (repositorio), ha agregado un repositorio y ha abierto el repositorio en el área de trabajo de Azure Databricks.
Cree un archivo denominado myfunctions.py en el repositorio y agregue el contenido siguiente al archivo. Otros ejemplos de este artículo esperan que este archivo se llame myfunctions.py. Puede usar nombres diferentes para sus propios archivos.
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
En el código siguiente se supone que ha configurado carpetas de Git de Databricks (repositorio), ha agregado un repositorio y ha abierto el repositorio en el área de trabajo de Azure Databricks.
Cree un archivo denominado myfunctions.r en el repositorio y agregue el contenido siguiente al archivo. Otros ejemplos de este artículo esperan que este archivo se llame myfunctions.r. Puede usar nombres diferentes para sus propios archivos.
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
Cree un cuaderno de Scala llamado myfunctions con el contenido siguiente: Otros ejemplos de este artículo esperan que este cuaderno se llame myfunctions. Puede usar nombres diferentes para sus propios cuadernos.
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
En el código siguiente se da por hecho que tiene los diamantes del conjunto de datos de ejemplo de terceros en un esquema denominado default dentro de un catálogo denominado main que es accesible desde el área de trabajo de Azure Databricks. Si el catálogo o esquema que quiere usar tiene un nombre diferente, cambie una o ambas instrucciones USE siguientes para que coincidan.
Cree un cuaderno de SQL y agregue el contenido siguiente a este cuaderno nuevo. Después, asocie el cuaderno a un clúster y ejecute el cuaderno para agregar las siguientes UDF de SQL al catálogo y el esquema especificados.
Nota:
Las UDF de SQL table_exists y column_exists solo funcionan con Unity Catalog. La compatibilidad con UDF de SQL para Unity Catalog está en versión preliminar pública.
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
Llamada a funciones
En esta sección se describe el código que llama a las funciones anteriores. Puede usar estas funciones, por ejemplo, para contar el número de filas de la tabla donde existe un valor especificado en una columna especificada. Pero, antes de continuar, le gustaría comprobar si la tabla existe realmente y también si la columna existe realmente en esa tabla. El código siguiente comprueba si estas condiciones son ciertas.
Si ha agregado las funciones de la sección anterior al área de trabajo de Azure Databricks, puede llamar a estas funciones desde el área de trabajo, tal como se indica a continuación.
Python
Cree un cuaderno de Python en la misma carpeta que el archivo anterior myfunctions.py en el repositorio y agregue el contenido siguiente al cuaderno. Cambie los valores de variable para el nombre de la tabla, el nombre del esquema (base de datos), el nombre de columna y el valor de columna según sea necesario. Después, asocie el cuaderno a un clúster y ejecute el cuaderno para ver los resultados.
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
Cree un cuaderno de R en la misma carpeta que el archivo anterior myfunctions.r en el repositorio y agregue el contenido siguiente al cuaderno. Cambie los valores de variable para el nombre de la tabla, el nombre del esquema (base de datos), el nombre de columna y el valor de columna según sea necesario. Después, asocie el cuaderno a un clúster y ejecute el cuaderno para ver los resultados.
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
Cree otro cuaderno de Scala en la misma carpeta que el cuaderno de Scala myfunctions anterior y agregue el contenido siguiente a este cuaderno nuevo.
En la primera celda de este nuevo cuaderno, agregue el código siguiente, que llama a la instrucción mágica %run. Esta instrucción mágica hace que el contenido del cuaderno myfunctions esté disponible para el nuevo cuaderno.
%run ./myfunctions
En la segunda celda de este cuaderno nuevo, agregue el código siguiente. Cambie los valores de variable para el nombre de la tabla, el nombre del esquema (base de datos), el nombre de columna y el valor de columna según sea necesario. Después, asocie el cuaderno a un clúster y ejecute el cuaderno para ver los resultados.
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
Agregue el código siguiente a una celda nueva del cuaderno anterior o a una celda de un cuaderno independiente. Cambie los nombres de esquema o catálogo si es necesario para que coincida con el suyo y, después, ejecute esta celda para ver los resultados.
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
Escritura de pruebas unitarias
En esta sección se describe el código que prueba cada una de las funciones que se describen al principio de este artículo. Si más adelante realiza cambios en las funciones, puede usar pruebas unitarias para determinar si esas funciones siguen funcionando según lo previsto.
Si ha agregado las funciones al principio de este artículo al área de trabajo de Azure Databricks, puede agregar pruebas unitarias para estas funciones al área de trabajo de la siguiente manera.
Python
Cree otro archivo llamado test_myfunctions.py en la misma carpeta que el archivo anterior myfunctions.py en el repositorio y agregue el contenido siguiente al archivo. De manera predeterminada, pytest busca los archivos .py cuyos nombres comiencen por test_ (o terminen por _test) para probar. Del mismo modo, de manera predeterminada, pytest busca funciones dentro de estos archivos cuyos nombres empiecen por test_ para probar.
En general, se recomienda no ejecutar pruebas unitarias en funciones que funcionan con datos en producción. Esto es especialmente importante para las funciones que agregan, quitan o cambian datos de otro modo. Para proteger sus datos de producción de ser comprometidos por sus pruebas unitarias de forma inesperada, debe ejecutar pruebas unitarias con datos que no están en producción. Un enfoque común es crear datos falsos lo más cerca posible de los datos de producción. En el ejemplo de código siguiente se crean datos falsos para que se ejecuten las pruebas unitarias.
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
Cree otro archivo llamado test_myfunctions.r en la misma carpeta que el archivo anterior myfunctions.r en el repositorio y agregue el contenido siguiente al archivo. De manera predeterminada, testthat busca los archivos .r cuyos nombres comiencen por test para probar.
En general, se recomienda no ejecutar pruebas unitarias en funciones que funcionan con datos en producción. Esto es especialmente importante para las funciones que agregan, quitan o cambian datos de otro modo. Para proteger sus datos de producción de ser comprometidos por sus pruebas unitarias de forma inesperada, debe ejecutar pruebas unitarias con datos que no están en producción. Un enfoque común es crear datos falsos lo más cerca posible de los datos de producción. En el ejemplo de código siguiente se crean datos falsos para que se ejecuten las pruebas unitarias.
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
Cree otro cuaderno de Scala en la misma carpeta que el cuaderno de Scala myfunctions anterior y agregue el contenido siguiente a este cuaderno nuevo.
En la primera celda de este nuevo cuaderno, agregue el código siguiente, que llama a la instrucción mágica %run. Esta instrucción mágica hace que el contenido del cuaderno myfunctions esté disponible para el nuevo cuaderno.
%run ./myfunctions
En la segunda celda, agregue el código siguiente. Este código define las pruebas unitarias y especifica cómo ejecutarlas.
En general, se recomienda no ejecutar pruebas unitarias en funciones que funcionan con datos en producción. Esto es especialmente importante para las funciones que agregan, quitan o cambian datos de otro modo. Para proteger sus datos de producción de ser comprometidos por sus pruebas unitarias de forma inesperada, debe ejecutar pruebas unitarias con datos que no están en producción. Un enfoque común es crear datos falsos lo más cerca posible de los datos de producción. En el ejemplo de código siguiente se crean datos falsos para que se ejecuten las pruebas unitarias.
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:
En este ejemplo de código se usa el estilo de pruebas FunSuite en ScalaTest. Para ver otros estilos de prueba disponibles, vea Selección de estilos de pruebas para el proyecto.
Sql
Antes de agregar pruebas unitarias, debe tener en cuenta que, en general, es un procedimiento recomendado no ejecutar pruebas unitarias en funciones que funcionan con datos en producción. Esto es especialmente importante para las funciones que agregan, quitan o cambian datos de otro modo. Para proteger sus datos de producción de ser comprometidos por sus pruebas unitarias de forma inesperada, debe ejecutar pruebas unitarias con datos que no están en producción. Un enfoque común es ejecutar pruebas unitarias en vistas en lugar de tablas.
Para crear una vista, puede llamar al comando CREATE VIEW desde una nueva celda del cuaderno anterior o desde un cuaderno independiente. En el ejemplo siguiente se supone que tiene una tabla existente denominada diamonds dentro de un esquema (base de datos) denominado default dentro de un catálogo denominado main. Cambie estos nombres para que coincidan con los suyos según sea necesario y, a continuación, ejecute solo esa celda.
USE CATALOG main;
USE SCHEMA default;
CREATE VIEW view_diamonds AS
SELECT * FROM diamonds;
Después de crear la vista, agregue cada una de las instrucciones SELECT siguientes a su propia celda nueva en el cuaderno anterior o a su propia celda nueva en un cuaderno independiente. Cambie los nombres para que coincidan con los suyos según sea necesario.
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'."));
Ejecute pruebas unitarias
En esta sección se describe cómo ejecutar las pruebas unitarias que ha codificado en la sección anterior. Al ejecutar las pruebas unitarias, obtendrá resultados que muestran las pruebas unitarias que se han superado y las que han dado errores.
Si ha agregado las pruebas unitarias de la sección anterior al área de trabajo de Azure Databricks, puede ejecutar estas pruebas unitarias desde el área de trabajo. Puede ejecutar estas pruebas unitarias de forma manual o en función de una programación.
Python
Cree un cuaderno de Python en la misma carpeta que el archivo anterior test_myfunctions.py en el repositorio y agregue el contenido siguiente.
En la primera celda del cuaderno nuevo, agregue el código siguiente y luego ejecute la celda, que llama a la instrucción mágica %pip. Esta instrucción mágica instala pytest.
%pip install pytest
En la segunda celda, agregue el código siguiente y luego ejecute la celda. En los resultados se muestran las pruebas unitarias que se han superado y las que han dado errores.
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
Cree un cuaderno de R en la misma carpeta que el archivo anterior test_myfunctions.r en el repositorio y agregue el contenido siguiente.
En la primera celda, agregue el código siguiente y luego ejecútela, lo que llama a la función install.packages. Esta función instala testthat.
install.packages("testthat")
En la segunda celda, agregue el código siguiente y luego ejecútela. En los resultados se muestran las pruebas unitarias que se han superado y las que han dado errores.
library(testthat)
source("myfunctions.r")
test_dir(".", reporter = "tap")
Scala
Ejecute la primera y, después, la segunda celda del cuaderno desde la sección anterior. En los resultados se muestran las pruebas unitarias que se han superado y las que han dado errores.
Sql
Ejecute cada una de las tres celdas del cuaderno desde la sección anterior. En los resultados se muestran si las pruebas unitarias se han superado o han dado errores.
Si ya no necesita la vista después de ejecutar las pruebas unitarias, puede eliminar la vista. Para eliminar esta vista, puede agregar el código siguiente a una nueva celda dentro de uno de los cuadernos anteriores y, a continuación, ejecutar solo esa celda.
DROP VIEW view_diamonds;
Sugerencia
Puede ver los resultados de las ejecuciones del cuaderno (incluidos los resultados de pruebas unitarias) en los registros de controladores del clúster. También puede especificar una ubicación para la entrega del registro del clúster.
Puede configurar un sistema de integración continua e implementación continua (CI/CD), como Acciones de GitHub, para ejecutar automáticamente las pruebas unitarias siempre que cambie el código. Para obtener un ejemplo, consulte la cobertura de Acciones de GitHub en Procedimientos recomendados de ingeniería de software para cuadernos.
Recursos adicionales
pytest
- Página principal de pytest
- Guías paso a paso de pytest
- Guías de referencia de pytest
- Procedimientos recomendados de ingeniería de software para cuadernos