Testing

Testing#

Have uv? ⚡

If you have uv installed, you can instantly open this page as a Jupyter notebook using opennb:

uvx --with "pipefunc[docs]" opennb pipefunc/pipefunc/docs/source/concepts/testing.md

This command creates an ephemeral environment with all dependencies and launches the notebook in your browser in 1 second - no manual setup needed! ✨.

Alternatively, run:

uv run https://raw.githubusercontent.com/pipefunc/pipefunc/refs/heads/main/get-notebooks.py

to download all documentation as Jupyter notebooks.

How to mock functions in a pipeline for testing?#

When mocking a function within a Pipeline for testing purposes, you can use the pipefunc.testing.patch utility. This is particularly useful for replacing the implementation of a function with a mock during tests, allowing you to control outputs and side effects.

Why not unittest.mock.patch?

The plain use of unittest.mock.patch is insufficient for Pipeline objects due to internal management of functions. Wrapped Functions: A Pipeline contains PipeFunc instances that store a reference to the original function in a func attribute. This structure means the function isn’t directly accessible by name for patching, as unittest.mock.patch would typically require.

See this example for how to use pipefunc.testing.patch to mock functions in a pipeline:

from pipefunc.testing import patch
from pipefunc import Pipeline, pipefunc, PipeFunc
import random

my_first = PipeFunc(random.randint, output_name="rnd", defaults={"a": 0, "b": 10})


@pipefunc(output_name="result")
def my_second(rnd):
    raise RuntimeError("This function should be mocked")


pipeline = Pipeline([my_first, my_second])

# Patch a single function
with patch(pipeline, "my_second") as mock:
    mock.return_value = 5
    print(pipeline())

# Patch multiple functions
with patch(pipeline, "random.randint") as mock1, patch(pipeline, "my_second") as mock2:
    mock1.return_value = 3
    mock2.return_value = 5
    print(pipeline())

5
5
Contents