Testing with PyTest

rkd.core.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: Running a task on a fully featured RKD executor

#!/usr/bin/env python3

import os
from rkd.core.api.testing import FunctionalTestingCase

SCRIPT_DIR_PATH = os.path.dirname(os.path.realpath(__file__))


class TestFunctional(FunctionalTestingCase):
    """
    Functional tests case of the whole application.
    Runs application like from the shell, captures output and performs assertions on the results.
    """

    def test_tasks_listing(self):
        """ :tasks """

        full_output, exit_code = self.run_and_capture_output([':tasks'])

        self.assertIn(' >> Executing :tasks', full_output)
        self.assertIn('[global]', full_output)
        self.assertIn(':version', full_output)
        self.assertIn('succeed.', full_output)
        self.assertEqual(0, exit_code)

Example: Mocking RKD-specific dependencies in TaskInterface

from rkd.core.api.inputoutput import BufferedSystemIO
from rkd.core.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())

API

class rkd.core.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'}):
    # code there

Parameters

environ –

Returns

static list_to_str ( in_list : list ) → List [ str ]

Execute __str__ on each list element, and replace element with the result

static mock_execution_context ( task : rkd.core.api.contract.TaskInterface , args : Optional [ Dict [ str , Union [ str , bool ] ] ] = None , env : Optional [ Dict [ str , str ] ] = None , defined_args : Optional [ Dict [ str , dict ] ] = None ) → rkd.core.api.contract.ExecutionContext

Prepares a simplified rkd.core.api.contract.ExecutionContext instance

Parameters

• task –

• args –

• env –

• defined_args –

Returns

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

Inserts required dependencies to your task that implements rkd.core.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.core.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.core.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

• task ( TaskInterface ) –

• args ( dict ) –

• env ( dict ) –

Returns

classmethod filter_out_task_events_from_log ( out : str )

Produces an array of events unformatted

[
    "Executing :sh -c echo 'Rocker' [part of :example]",
    "Executing :sh -c echo 'Kropotkin' [part of :example]",
    'Executing :sh -c echo "The Conquest of Bread"; exit 1 [part of :example]',
    'Retrying :sh -c echo "The Conquest of Bread"; exit 1 [part of :example]',
    'Executing :sh -c exit 0 ',
    'Executing :sh -c echo "Modern Science and Anarchism"; [part of :example]'
]

Parameters

out –

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

with_temporary_workspace_containing ( files : Dict [ str , str ] )

Creates a temporary directory as a workspace and fills up with files specified in “file” parameter

Parameters

files – Dict of [filename: contents to write to a file]

Returns

class rkd.core.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.