Skip to main content

Helps testing Kedro plugin.

Project description


GitHub Workflow Status GitHub Workflow Status Codecov PyPI

Helps you testing Kedro plugins with pytest.

from kedro_pytest import TestKedro

def test_count_cli(tkedro: TestKedro):'project')
    # ^ creates a new project in a temporary directory
    # ^ ceates a pipeline and two csv entries in catalog by default
        {'new_dset': {'type': 'pandas.JSONDataSet', 'filepath': 'a.json'}})
    # ^ adds a new dataset to the catalog
    result = tkedro.cli(['count', 'catalog'], ['--type', 'pandas.CSVDataSet'])
    # ^ returns a click.testing.Result object
    assert int(result.output) == 2
    # ^ cleans the temporary directory (only required for doctests)

def test_helloworld_hook(tkedro: TestKedro):
    with'project') as tk:
      assert 'Hello world!' in'pipe').output

This plugin focuses on generating minimal Kedro structures, such as small pipelines, and a small project, in order to simplify testing of Kedro plugins in a real Kedro environment. Also, to make it safe for testing, the plugin relies on pytest-tmpfs to create a temporary filesystem for the project.


For installing this package, just type the code below in your terminal:

pip install kedro-pytest


As a fixture

You can use the KedroPytest as a fixture by passing tkedro as an argument of your test function.

from kedro_pytest import TestKedro

def test_fixture(tkedro: TestKedro):

In doctests

You can always use fixtures in doctests. To do so, you have to inject it into the doctest namespace in the conftest file.

def add_to_doctests(doctest_namespace: dict, tkedro: TestKedro):
    """Adds the tkedro fixture to the doctest namespace."""
    doctest_namespace["kedro"] = tkedro

def function():
      >>> kedro.create_pipeline('pipe')
      >>> kedro.stop()

Your own instance

If you seek to use this plugin without pointing to tmp_path it is possible to create your own instance and manipulate it the way you want.

from kedro_pytest import TestKedro
from pathlib import Path

def test_my_path():
    tkedro = TestKedro(path=Path.cwd())


The following table contains all methods that TestKedro implements and its descriptions:

Method Description Parameters Return
new Creates a new project in a temporary directory. name: str TestKedro
create_pipeline Creates a minimal dummy pipeline with two csv entries in catalog by default and a parameter. name: str, *content: str ~
write_yml Writes a yaml file in the project. path: str, content: dict str: abs path to the file
read_yml Reads a yaml file in the project. path: str dict: content of the file
update_yml Updates a yaml file in the project. Can be nested, it will replace recursively. path: str, content: dict str: abs path to the file
cli Runs a kedro cli command in the project. command: list[str], args: list[str] click.testing.Result
run Runs a pipeline in the project. pipeline: str, run_command: list[str] click.testing.Result
stop Cleans the temporary directory. ~ ~

Note The create_pipeline method differs from the cli command because it creates a minimal pipeline (a with a dummy implementation, a catalog.yml, and a parameters.yml), and it automatically adds the pipeline to the registry, while the cli create_pipeline would create a folder containing a clean pipeline skeleton and will not add it to the registry.

Warning cwd switches between temporary and the last cwd when using new and stop methods of TestKedro. A new instance of the fixture is created for each test, so you don't have to worry about it in tests, but in doctests it may be useful. Also, the use of stop can be avoided by using with statements.

Advanced usage

If the methods mentioned above are not enough for your needs, you can always use the fs attribute of the TestKedro instance to manipulate the filesystem.

import json

def test_advanced(tkedro: TestKedro):'project')
    tkedro.cli(['json', 'catalog'])
    text = tkedro.fs.read_text('conf/base/catalog.json')


Feel free to contribute to this project, just remember to install pre-commit, and to write unit tests for it.

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

kedro_pytest-0.1.3-py3-none-any.whl (12.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page