Unit Testing Stages

Let us add some basic system tests to the ExampleConsumer class describe earlier, so that we can make sure our stage is behaving as expected while we work on it.

Kotekan uses pytest for its system tests. Through the pytest framework you can set up a corresponding producer stage/input buffer, automate the creation of a pipeline configuration, run your consumer stage, and then make assertions on its behaviour.

The steps for building a test are:

1. Set the parameters for the configuration of your stage.

2. Write a pytest fixture that runs kotekan:

   • Create a temporary directory for the data produced by your test.

   • If the stage you are testing has consumer components, set up a producer stage which generates the input buffer.

   • If the stage you are testing has producer components, set up the data dumper, which will allow you to validate your stage’s behaviour.

   • Configure KotekanStageTester with your stage, and tell it where it will can find the input buffer, and/or where it can output its buffer.

   • yield the final buffer for asserting.

3. Create the actual test. Pass it your stage’s pytest fixture. Assert the expected output frames.

import pytest
from kotekan import runner

# this is the equivalent of the pipeline config file for kotekan to run your test
params = {
    "num_elements": 7,
    "num_ev": 0,
    "total_frames": 128,
    "cadence": 10.0,
    "mode": "default",
    "variable_my_stage_needs": -1,
}

# this runs kotekan and yields the data you want to inspect
@pytest.fixture(scope="module")
def data(tmpdir_factory):

    # keep all the data this test produces in a tmp directory
    tmpdir = tmpdir_factory.mktemp("name_of_the_test_case")

    # you can use FakeVisBuffer as a mock producer stage
    # to produce a Kotekan buffer with fake data
    fakevis_buffer = runner.FakeVisBuffer(
        num_frames=params["total_frames"], mode=params["mode"]
    )

    # DumpVisBuffer can be used to dump data
    # afterwards for asserting expectations
    dump_buffer = runner.DumpVisBuffer(str(tmpdir))

    # KotekanStageTester is used to run your
    # kotekan stage with your config
    # Replace stage_type with the name of your stage
    test = runner.KotekanStageTester(
        stage_type="stageUnderTest",
        stage_config={},
        buffers_in=fakevis_buffer,
        buffers_out=dump_buffer,
        global_config=params,
    )

    test.run()

    # here the data that the stage under test
    # put out is passed on to test the stage
    yield dump_buffer.load()


# this is the actual test (give a name to it)
def test_<name>(data):

    for frame in data:
        assert frame.vis == {1, 0}

And there you go!

To run the test, while in the kotekan repo, type:

kotekan$ export PYTHONPATH=python
kotekan$ pytest -sv location_of_test/example_test.py

If it cannot find your new stage, ensure that pytest is running the correct local kotekan build, which includes your new code.

And congratulations! You now have the foundation in place to further develop your stage.