Module panama.utils.mock_utils

This module handles the mocking for Panama and the other Sorgenia projects. It's intended to be used in unit and integration tests, and allows the mocking of functions, methods, and the removal of code that should not be run during testing, for example scala code used to write data.

An example of how this code should work is the following: in a databricks notebook, run the following code.

from panama.utils.mock_utils import remove_scala_code
from panama.utils.nbutils import Notebook

# set input and output destination
notebook_to_test = "/path/to/notebook/to/test"
destination_notebook = "/path/to/notebook/copy/without/scala"

# create a notebook copy without scala
remove_scala_code(input_path=notebook_to_test, output_path=destination_notebook)

# notebook to test
nb_to_test = Notebook.read_notebook(destination_notebook)

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

from panama.utils.mock_utils import generate_mocks

# mocking stage

replacement_func = lambda **x: x["data"].write.format("csv").mode("overwrite").option("header", True).save("/path/to/test.csv")

patches = {
    "panama.io.IOAdls.read_file": {
        "fist_table_to_mock": {
            "arg_name": <name of the parameter used to flag a mocking situation>,
            "arg_value": <value of the parameter that triggers the mock>,
            "mock_value": <value to return>
        },

        "second_table_to_mock": {
            "arg_name": <name of the parameter used to flag a mocking situation>,
            "arg_value": <value of the parameter that triggers the mock>,
            "mock_value": <value to return>
        },

    },
    "panama.io.IOAdls.write_file": {
        "arg_name": <name of the value used to flag the mocking situation>,
        "arg_value": <value of the parameter that triggers the mock>,
        "mock_value": replacement_func
    }
}

generate_mocks(patches)

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

%run "/path/to/notebook/copy/without/scala"

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

# testing stage
from chispa.dataframe_comparer import assert_df_equality

# here tests must be run
truth = spark.table("/path/to/production").where("<filters>")
test = spark.table("/path/to/test.csv")

assert_df_equality(truth, test)

Functions

def generate_mocks(mocks: Dict[str, Any], required_keys: List[str] = ['arg_name', 'arg_value', 'mock_value'])

Function used to generate the mocking based on the mocks dictionary passed during initialization. The dictionary passed must have the following structure:

{ ".[\.]\>": { "": { "arg_name": "", "arg_value": "", "mock_value": "", }, }, }

An error is raised if any of these requirements is not satisfied. The mock_value entry can be either a fixed value or a function. If a function is passed, it will be executed with the same parameters of the method that will be mocked. Keep that in mind when creating the required function. An example of function that can be used to replace a writing function that takes the dataframe to write in an argument named 'data' is the following:

replacement_func = lambda **x: x["data"].write.format("csv").mode("overwrite").option("header", True).save("/path/to/test.csv")

Example:

{ "panama.io.IOAdls.read_most_recent_file": { "ptrasf_association": { "arg_name": "file_name", "arg_value": "PTrasf_Association", "mock_value": spark.createDataFrame([["a", 1], ["b", 2]]), }, }, "panama.utils.spark_utils.runtime": { "ptrasf_association": { "arg_name": "file_name", "arg_value": "PTrasf_Association", "mock_value": "local", }, }, }

Args

mocks : Dict[str, Any]

dictionary of the mock parameters.

required_keys : List[Union[str, List[str]]], optional

list of required items in the mock dict. Defaults to ["arg_name", "arg_value", ["mock_value", "mock_function"]].

def mock_method(method: Callable, mock_dict: Dict[str, Any]) ‑> Callable

This method mocks another method based on the mocks dictionary passed in. The dictionary is expected to be in the following format: { "mock_1": { "arg_name": , "arg_value": , "mock_value": } }.

More than one mock case can be passed.

Args

method : Callable

the method to be mocked.

mock_dict : Dict[str, Any]

the dict holding the mocking specifications.

Returns

Callable

the mocked method.

def parse_path(path: str) ‑> Tuple[Optional[str], Optional[str], str]

Function used to parse the path, considering that a method might not belong to a class, but a module.

Args

path : str

path to parse.

Returns

Tuple[str, Union[str, None], str]

path parts: module, class (if present), method.

def remove_scala_code(input_path: str, output_path: str) ‑> None

Function used to load a file, remove scala content, and write the new content to a file. It will only remove the cells that have a 'scala' magic. If the file is a scala databricks notebook, no operations can be performed.

Args

input_path : str

path of the file to read.

output_path : str

path of the file to write.