TestRig#

class maize.utilities.testing.TestRig(cls: type[Node], config: Config | None = None)[source]#

Bases: object

Test rig for user Node tasks. Can be loaded with parameters and inputs.

Parameters:
  • clsNode child class to be wrapped

  • config – Global configuration for the parent workflow

inputs#

Dictionary of input values

Type:

dict[str, Any]

parameters#

Dictionary of parameter values

Type:

dict[str, Any]

node#

Node instance

Type:

maize.core.node.Node

See also

MockChannel

Class simulating the behaviour of normal channels without the issues associated with multiple processes

Examples

>>> rig = TestRig(Foo)
... rig.set_inputs(inputs=dict(inp="bar"))
... rig.set_parameters(parameters=dict(param=42))
... rig.setup(n_outputs=2)
... outputs = rig.run()
__init__(cls: type[Node], config: Config | None = None) None[source]#

Methods

__init__(cls[, config])

run()

Run the node with inputs and parameters previously set.

set_inputs(inputs)

Set the task input values

set_parameters(parameters)

Set the task parameter values

setup([n_outputs])

Instantiate the node and create mock interfaces.

setup_run([inputs, parameters, n_outputs])

Instantiate and run the node with a specific set of parameters and inputs.

Attributes

run() None[source]#

Run the node with inputs and parameters previously set.

set_inputs(inputs: dict[str, Any] | None) None[source]#

Set the task input values

set_parameters(parameters: dict[str, Any] | None) None[source]#

Set the task parameter values

setup(n_outputs: int | dict[str, int] | None = None, **kwargs: Any) dict[str, MockChannel[Any]][source]#

Instantiate the node and create mock interfaces.

setup_run(inputs: dict[str, Any] | None = None, parameters: dict[str, Any] | None = None, n_outputs: int | dict[str, int] | None = None, **kwargs: Any) dict[str, MockChannel[Any]][source]#

Instantiate and run the node with a specific set of parameters and inputs.