Testing with unittest

rkd.api.testing provides methods for running tasks with output capturing, a well as mocking RKD classes for unit testing of your task methods. To use our API just extend one of base classes.

Example: Mocking RKD-specific dependencies in TaskInterface

from rkd.api.inputoutput import BufferedSystemIO
from rkd.api.testing import FunctionalTestingCase

# ...

class SomeTestCase(FunctionalTestingCase):

    # ...

    def test_something_important(self):
        task = LineInFileTask()  # put your task class there
        io = BufferedSystemIO()

        BasicTestingCase.satisfy_task_dependencies(task, io=io)

        self.assertEqual('something', task.some_method())

Documentation

class rkd.api.testing.BasicTestingCase(methodName='runTest')
Provides minimum of:
  • Doing backup of environment and cwd
  • Methods for mocking task dependencies (RKD-specific like ExecutionContext)
environment(environ: dict)

Mocks environment

Example usage:
with self.environment({‘RKD_PATH’: SCRIPT_DIR_PATH + ‘/../docs/examples/env-in-yaml/.rkd’}):
Parameters:environ
Returns:
static mock_execution_context(task: rkd.api.contract.TaskInterface, args: Dict[str, str] = None, env: Dict[str, str] = None, defined_args: Dict[str, dict] = None) → rkd.api.contract.ExecutionContext

Prepares a simplified rkd.api.contract.ExecutionContext instance

Parameters:
  • task
  • args
  • env
  • defined_args
Returns:

static satisfy_task_dependencies(task: rkd.api.contract.TaskInterface, io: rkd.api.inputoutput.IO = None) → rkd.api.contract.TaskInterface

Inserts required dependencies to your task that implements rkd.api.contract.TaskInterface

Parameters:
  • task
  • io
Returns:

setUp() → None

Hook method for setting up the test fixture before exercising it.

tearDown() → None

Hook method for deconstructing the test fixture after testing it.

class rkd.api.testing.FunctionalTestingCase(methodName='runTest')

Provides methods for running RKD task or multiple tasks with output and exit code capturing. Inherits OutputCapturingSafeTestCase.

execute_mocked_task_and_get_output(task: rkd.api.contract.TaskInterface, args=None, env=None) → str

Run a single task, capturing it’s output in a simplified way. There is no whole RKD bootstrapped in this operation.

Parameters:
Returns:

run_and_capture_output(argv: list, verbose: bool = False) → Tuple[str, int]

Run task(s) and capture output + exit code. Whole RKD from scratch will be bootstrapped there.

Example usage:
full_output, exit_code = self.run_and_capture_output([‘:tasks’])
Parameters:
  • argv (list) – List of tasks, arguments, commandline switches
  • verbose (bool) – Print all output also to stdout
Returns:

class rkd.api.testing.OutputCapturingSafeTestCase(methodName='runTest')

Provides hooks for keeping stdout/stderr immutable between tests.

setUp() → None

Hook method for setting up the test fixture before exercising it.

tearDown() → None

Hook method for deconstructing the test fixture after testing it.