10-31-2022 12:43 PM
11-21-2022 07:30 AM
Hi,
You can mock dbutils. An easy way to do it is to:
Example:
def my_function(dbutils):
...
dbutils.fs.ls("/tmp") # this is using the local variable received by parameter
...
from my_library import my_function
...
my_function(dbutils) # this refers to the global dbutils variable
from unittest.mock import MagicMock
from my_library import my_function
def test_my_function_calls_dbutils():
mock_dbutils = MagicMock()
my_function(dbutils=mock_dbutils)
mock_dbutils.fs.ls.assert_called_once_with("/tmp")
11-21-2022 07:40 AM
above reponse is a pure mock, and below is another example from dbx about a fixture with local filesystem, you can also add mock on dbutils.secrets:
01-15-2023 10:13 PM
Hi @Jake P
Hope all is well! Just wanted to check in if you were able to resolve your issue and would you be happy to share the solution or mark an answer as best? Else please let us know if you need more help.
We'd love to hear from you.
Thanks!
10-24-2024 04:36 AM
How can I add locally type annotation to the example provided above by @fermin_vicente?
The type of `dbutils` seems to be `dbruntime.dbutils.DBUtils` but I'm not sure how to get it available in local dev.
10-24-2024 10:57 PM
I personally went without type hint for the dbutils param to avoid extra dependencies just for that, but I think you can actually get that type from the Databricks SDK, according to the docs:
https://docs.databricks.com/en/dev-tools/databricks-connect/python/databricks-utilities.html
https://pypi.org/project/databricks-sdk/#description
I would worry in this case about matching the proper SDK with the DBR you're using to ensure the type validations match (although I guess dbutils doesn't evolve a lot)
Hope this helps
10-24-2024 11:56 PM
import unittest
from unittest.mock import MagicMock, patch
# Import the function to be tested
from your_notebook import my_function
class TestMyFunction(unittest.TestCase):
@patch('your_notebook.dbutils')
def test_my_function(self, mock_dbutils):
# Create a mock for dbutils.fs
mock_fs = MagicMock()
mock_dbutils.fs = mock_fs
# Define the behavior of the mock methods
mock_fs.mkdirs.return_value = None
mock_fs.ls.return_value = ["file1", "file2"]
# Call the function to be tested
result = my_function()
# Assertions
if __name__ == '__main__':
unittest.main()
10-25-2024 02:44 AM
Fermin_vicente's answer is pretty good already. Below is how you can do something similar with conftest.py
# conftest.py
import pytest
from unittest.mock import MagicMock
from pyspark.sql import SparkSession
@pytest.fixture(scope="session")
def dbutils():
dbutils = MagicMock()
yield dbutils
dbutils.stop()
# Also for the SparkSession
@pytest.fixture(scope="session")
def spark():
spark = (
SparkSession.builder.master("local[*]").appName("app-name").getOrCreate()
)
yield spark
spark.stop()
Then for the testing code:
def test_query_generation(spark, dbutils):
"""
Test example
"""
test_obj = QueryGenerator(test_config, spark, dbutils)
expected_query = (
"SELECT * from community.databricks.com"
)
assert (
test_obj.get_query() == expected_query
)
Your class object has to get input a dbutils object. I think the same can be done if you want to use just functions.
Join a Regional User Group to connect with local Databricks users. Events will be happening in your city, and you won’t want to miss the chance to attend and share knowledge.
If there isn’t a group near you, start one and help create a community that brings people together.
Request a New Group