Testing a Processing Step

While it is possible to invoke a ProcessingStep from the command line, ProcessingSteps can also be called from within Python for local unit testing. This test interface is currently under development, and is subject to change.

In order to test a ProcessingStep, we must create an instance of our ProcessingStep, and invoke a function by name. Function parameters must be passed in as a dict; DataSlots are passed in via create_dataslot_description(), which takes a dict from argument names to lists of filenames. Note that output DataSlots must also be specified in the same way as input DataSlots. Result DataSlots are a special case: while they do not need to be specified, you will likely wish to test that the results serialize correctly. The example below uses a pytest fixture (not shown) to generate a temporary output file for the test.

test_cloud_processing.py

import os
from conftest import create_tmp_file

from cloud_step import CloudProcessingStep, CloudRegions
from models import Region
from pinexq.procon.dataslots import create_dataslot_description
from pinexq.procon.dataslots.annotation import RETURN_SLOT_NAME #=="__returns__"
from pinexq.procon.step import ExecutionContext

def test_get_job_results(create_tmp_file):
    """
    Test retrieving jobs from Cloud Service.
    """
    worker = CloudProcessingStep(use_cli=False) # Suppress pop-up
    tmp_output_file = create_tmp_file()
    result = worker._call(
        ExecutionContext(
            function_name="get_task_results",
            input_dataslots=create_dataslot_description(
                {"license_file": [os.environ["CLOUD_LICENSE_PATH"]]} # load license file from environmental variable-specified path
            ),
            parameters={
                "region": Region(CloudRegions.EU_WEST_2),
                "backend_name": "cb1",
                "job_id": "e01200e5-c000-4003-9000-c0abb7000c52"
            },
            output_dataslots=create_dataslot_description(
                {RETURN_SLOT_NAME: [str(tmp_output_file)]} # write result to temp file
            )
        )
    )
    # "result" is function return value if return DataSlot is not specified. Otherwise...
    assert result is None
    result_json = tmp_output_file.read_text()
    assert result_json is not None

Some ProcessingSteps may require API keys in order to test properly, especially those accessing external resources. We strongly recommend you do not include these API keys in your source code; see the discussion here for alternatives.

Mocking execution context

In certain cases, it may be necessary to mock various methods your ProcessingStep calls, especially if those methods depend on being called by the JMA. For example, if your function acquires a Client from its step context, you will need to supply a Client during testing. An example is shown below, using pytest-mock to mock getting the Client, and creating the Client itself with a fixture.

from main import UploadStep
from pinexq.procon.step import ExecutionContext

def test_sync_step(client, mocker):
    """
    Test a step that uploads workdata.
    Mocks get_client() with mocker, supplying an API client via fixture.
    """
    step = UploadStep(use_cli=False)
    # Note: patch where function is looked up, not where it's defined.
    # See e.g. https://docs.python.org/3/library/unittest.mock.html#where-to-patch.
    mocker.patch("main.get_client", return_value=client)
    step._call(
        ExecutionContext(
            function_name="sync_files"
        )
    )

Debugging on a remote JMA

In more complicated cases, it may be necessary to debug your worker while it processes a job on our JMA. By authenticating with a remote-enabled API key, and setting environment variables and launch arguments in your IDE, it is possible to debug a worker in remote mode; you can then step through your code, line by line, and watch the stack as it executes. For more information, see Using ProCon from CLI.

Note that, in order to ensure that your worker receives a particular job, it may be necessary to scale down deployment of workers implementing the same function; alternatively, you may wish to register a new pre-release version for testing.