Блог

Я создал несколько классов, каждый из которых принимает dataframe в качестве параметра. Я импортировал pytest и создал несколько приспособлений и простых методов assert.

Я могу вызвать pytest.main([.]) из записной книжки, и он выполнит pytest из корневого каталога (databricks / driver).

Я попытался передать путь к записной книжке, но он говорит, что не найден.

В идеале я бы хотел выполнить это из командной строки.

Как мне настроить корневой каталог? Похоже, существует разрыв между spark os и областью пользовательского рабочего пространства, который мне трудно подключить.

В качестве предостережения я не хочу использовать unittest, поскольку i pytest может быть успешно использован в CI pipleine путем вывода junitxml, о котором AzureDevOps может сообщать.

Я объяснил причину, по которой вы не можете запускать pytest в записных книжках Databricks (если вы не экспортируете их и не загружаете в dbfs в виде обычных файлов .py, что не то, что вам нужно) по ссылке внизу этого поста.

Тем не менее, я смог запускать doctests в Databricks, используя doctest.run_docstring_examples метод, подобный этому:

 import doctest

def f(x):
    """
    >>> f(1)
    45
    """
    return x   1

doctest.run_docstring_examples(f, globals())
  

Это распечатает:

 **********************************************************************
File "/local_disk0/tmp/1580942556933-0/PythonShell.py", line 5, in NoName
Failed example:
    f(1)
Expected:
    45
Got:
    2
  

Если вы также хотите вызвать исключение, взгляните подробнее на: https://menziess.github.io/howto/test/code-in-databricks-notebooks

Взято из собственного репозитория Databricks: https://github.com/databricks/notebook-best-practices/blob/main/notebooks/run_unit_tests.py

 # Databricks notebook source
# MAGIC %md Test runner for `pytest`

# COMMAND ----------

!cp ../requirements.txt ~/.
%pip install -r ~/requirements.txt

# COMMAND ----------

# pytest.main runs our tests directly in the notebook environment, providing
# fidelity for Spark and other configuration variables.
#
# A limitation of this approach is that changes to the test will be
# cache by Python's import caching mechanism.
#
# To iterate on tests during development, we restart the Python process 
# and thus clear the import cache to pick up changes.
dbutils.library.restartPython()

import pytest
import os
import sys

# Run all tests in the repository root.
notebook_path = dbutils.notebook.entry_point.getDbutils().notebook().getContext().notebookPath().get()
repo_root = os.path.dirname(os.path.dirname(notebook_path))
os.chdir(f'/Workspace/{repo_root}')
%pwd

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

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

# Fail the cell execution if we have any test failures.
assert retcode == 0, 'The pytest invocation failed. See the log above for details.'