pytest-pipeline 0.1.0

Installation

pip install pytest-pipeline

Walkthrough

For our example, we will use a super simple pipeline that writes a file and prints to stdout:

#!/usr/bin/env python

from __future__ import print_function

if __name__ == "__main__":

    with open("result.txt", "w") as result:
        result.write("42\n")
    print("Result computed")

At this point it’s just a simple script, but it should be enough to illustrate the plugin. Also, if you want to follow along, save the above file as run_pipeline.

With the pipeline above, here’s how your test would look like with pytest_pipeline:

import os
import shutil
from pytest_pipeline import PipelineRun, PipelineTest, mark, utils

# one pipeline run is represented by one class that subclasses PipelineTest
class TestMyPipeline(PipelineTest):

    # define the pipeline execution via PipelineRun objects
    run = PipelineRun(
        # the actual command to start your pipeline
        cmd="./run_pipeline",
        stdout="run.stdout",
    )

    # before_run-marked functions will be run before the pipeline is executed
    @mark.before_run
    def test_prep_executable(self):
        # copy the executable to the run directory
        shutil.copy2("/path/to/run_pipeline", "run_pipeline")
        # testing if the file is executable
        assert os.access("run_pipeline", os.X_OK)

    # after_run-marked tests will only be run after pipeline execution is finished
    @mark.after_run(order=1)
    def test_result_md5(self):
        assert utils.file_md5sum("result.txt") == "50a2fabfdd276f573ff97ace8b11c5f4"

    # ordering for all tests annotated by after_run can be set manually
    # here we want to test the exit code first after the run is finished
    @mark.after_run(order=0)
    def test_exit_code(self):
        assert self.run.exit_code == 0

    # we can also check the stdout that we capture as well
    @mark.after_run(order=2)
    def test_stdout(self):
        assert open("run.stdout", "r").read().strip() == "Result computed"

If the test above is saved as test_demo.py, you can then run the test by executing py.test -v test_demo.py. You should see that four tests were executed and all four passed.

Getting + giving help

Please use the issue tracker to report bugs or feature requests. You can always fork and submit a pull request as well.

License

See LICENSE.