Databricks Test Case

Every notebook tests should inherit from DatabricksTestCase.

Source code in databricks_testing_tools/test_case.py

class DatabricksTestCase(unittest.TestCase):
    """
    Every notebook tests should inherit from DatabricksTestCase.
    """

    def execute_tests(self, output: str = "", run_in_parallel: Union[bool, int] = False) -> Any:
        """
        Execute tests and write results to a directory or stdout. 
        Tests can optionally be run in parallel to speed up execution.

        Args:
            output (str): The directory to write the results to. If empty, results will be written to stdout.
            run_in_parallel (Union[bool, int]): If False (default), tests are run sequentially. 
                                                 If True, as many tests are run in parallel as there are available.
                                                 If an integer is provided, that many tests will be run in parallel.

        Returns:
            Any: The test case results. When tests are run in parallel, this will be a list of results.

        Note:
            When running tests in parallel, make sure the tests are designed to be able to run concurrently.
            Tests that are not safe to run in parallel could have unexpected results or cause errors.
        """
        loader = unittest.TestLoader()
        suite = loader.loadTestsFromTestCase(type(self))
        results = []

        if output == "":
            runner = unittest.TextTestRunner(stream=sys.stdout)
        else:
            runner = xmlrunner.XMLTestRunner(
                output=output.rstrip("/") + "/junit/"
            )

        if run_in_parallel:
            test_case_count = suite.countTestCases()
            worker_number = test_case_count if isinstance(run_in_parallel, bool) else min(test_case_count, run_in_parallel)

            with ThreadPoolExecutor(max_workers=worker_number) as executor:
                results = list(executor.map(lambda test: runner.run(unittest.TestSuite([test])), suite))
        else:
            results = runner.run(unittest.TestSuite(suite))

        return results

execute_tests(self, output: str = '', run_in_parallel: Union[bool, int] = False) -> Any

Execute tests and write results to a directory or stdout. Tests can optionally be run in parallel to speed up execution.

Parameters:

Name	Type	Description	Default
output	str	The directory to write the results to. If empty, results will be written to stdout.	''
run_in_parallel	Union[bool, int]	If False (default), tests are run sequentially. If True, as many tests are run in parallel as there are available. If an integer is provided, that many tests will be run in parallel.	False

Returns:

Type	Description
Any	The test case results. When tests are run in parallel, this will be a list of results.

!!! note When running tests in parallel, make sure the tests are designed to be able to run concurrently. Tests that are not safe to run in parallel could have unexpected results or cause errors.

Source code in databricks_testing_tools/test_case.py

def execute_tests(self, output: str = "", run_in_parallel: Union[bool, int] = False) -> Any:
    """
    Execute tests and write results to a directory or stdout. 
    Tests can optionally be run in parallel to speed up execution.

    Args:
        output (str): The directory to write the results to. If empty, results will be written to stdout.
        run_in_parallel (Union[bool, int]): If False (default), tests are run sequentially. 
                                             If True, as many tests are run in parallel as there are available.
                                             If an integer is provided, that many tests will be run in parallel.

    Returns:
        Any: The test case results. When tests are run in parallel, this will be a list of results.

    Note:
        When running tests in parallel, make sure the tests are designed to be able to run concurrently.
        Tests that are not safe to run in parallel could have unexpected results or cause errors.
    """
    loader = unittest.TestLoader()
    suite = loader.loadTestsFromTestCase(type(self))
    results = []

    if output == "":
        runner = unittest.TextTestRunner(stream=sys.stdout)
    else:
        runner = xmlrunner.XMLTestRunner(
            output=output.rstrip("/") + "/junit/"
        )

    if run_in_parallel:
        test_case_count = suite.countTestCases()
        worker_number = test_case_count if isinstance(run_in_parallel, bool) else min(test_case_count, run_in_parallel)

        with ThreadPoolExecutor(max_workers=worker_number) as executor:
            results = list(executor.map(lambda test: runner.run(unittest.TestSuite([test])), suite))
    else:
        results = runner.run(unittest.TestSuite(suite))

    return results