aiida.backends.tests.engine package

Submodules

Test for the CalcJob process sub class.

class aiida.backends.tests.engine.test_calc_job.TestCalcJob(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test for the CalcJob process sub class.

__module__ = 'aiida.backends.tests.engine.test_calc_job'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_run_base_class()[source]

Verify that it is impossible to run, submit or instantiate a base CalcJob class.

Tests for the calcfunction decorator and CalcFunctionNode.

class aiida.backends.tests.engine.test_calcfunctions.TestCalcFunction(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for calcfunctions.

__module__ = 'aiida.backends.tests.engine.test_calcfunctions'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_calcfunction_caching()[source]

Verify that a calcfunction can be cached.

test_calcfunction_caching_change_code()[source]

Verify that changing the source codde of a calcfunction invalidates any existing cached nodes.

test_calcfunction_default_linkname()[source]

Verify that a calcfunction that returns a single Data node gets a default link label.

test_calcfunction_do_not_store_provenance()[source]

Run the function without storing the provenance.

Verify that a calcfunction can only get CREATE links and no RETURN links.

test_calcfunction_node_type()[source]

Verify that a calcfunction gets a CalcFunctionNode as node instance.

test_calcfunction_return_stored()[source]

Verify that a calcfunction will raise when a stored node is returned.

test_calculation_call_store_provenance_false()[source]

Verify that a calcfunction can call another calcfunction as long as store_provenance is False.

test_calculation_cannot_call()[source]

Verify that calling another process from within a calcfunction raises as it is forbidden.

class aiida.backends.tests.engine.test_class_loader.TestCalcJob(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_class_loader'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_class_loader()[source]
class aiida.backends.tests.engine.test_daemon.TestDaemon(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_daemon'
class aiida.backends.tests.engine.test_futures.TestWf(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

TIMEOUT = datetime.timedelta(0, 5)
__module__ = 'aiida.backends.tests.engine.test_futures'
test_calculation_future_broadcasts()[source]
test_calculation_future_polling()[source]
class aiida.backends.tests.engine.test_launch.AddWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_launch'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
add()[source]
classmethod define(spec)[source]
class aiida.backends.tests.engine.test_launch.TestLaunchers(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_launch'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_calcfunction_run()[source]
test_calcfunction_run_get_node()[source]
test_calcfunction_run_get_pid()[source]
test_workchain_builder_run()[source]
test_workchain_builder_run_get_node()[source]
test_workchain_builder_run_get_pid()[source]
test_workchain_run()[source]
test_workchain_run_get_node()[source]
test_workchain_run_get_pid()[source]
aiida.backends.tests.engine.test_launch.add(*args, **kwargs)[source]
class aiida.backends.tests.engine.test_persistence.TestAiiDAPersister(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_persistence'
maxDiff = 1024
setUp()[source]

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

test_delete_checkpoint()[source]
test_save_load_checkpoint()[source]
class aiida.backends.tests.engine.test_persistence.TestProcess(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the basic saving and loading of process states.

__module__ = 'aiida.backends.tests.engine.test_persistence'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_save_load()[source]
class aiida.backends.tests.engine.test_process.NameSpacedProcess(inputs=None, logger=None, runner=None, parent_pid=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.process.Process

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_process'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
_node_class

alias of aiida.orm.nodes.process.workflow.workflow.WorkflowNode

classmethod define(spec)[source]
class aiida.backends.tests.engine.test_process.ProcessStackTest(inputs=None, logger=None, runner=None, parent_pid=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.process.Process

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_process'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
_node_class

alias of aiida.orm.nodes.process.workflow.workflow.WorkflowNode

on_create()[source]
on_stop()[source]
run()[source]
class aiida.backends.tests.engine.test_process.TestProcess(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_process'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_description()[source]
test_inputs()[source]
test_label()[source]
test_none_input()[source]
test_process_stack()[source]
test_process_type_with_entry_point()[source]

For a process with a registered entry point, the process_type will be its formatted entry point string

test_process_type_without_entry_point()[source]

For a process without a registered entry point, the process_type will fall back on the fully qualified class name

test_save_instance_state()[source]
test_seal()[source]
test_validation_error()[source]

Test that validating a port produces an meaningful message

test_work_calc_finish()[source]
class aiida.backends.tests.engine.test_process.TestProcessNamespace(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_process'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_namespaced_process()[source]

Test that inputs in nested namespaces are properly validated and the link labels are properly formatted by connecting the namespaces with underscores

class aiida.backends.tests.engine.test_process_builder.TestProcessBuilder(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_process_builder'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_code_get_builder()[source]

Test that the get_builder method of Code returns a builder where the code is already set.

test_dynamic_getters_doc_string()[source]

Verify that getters have the correct docstring

test_dynamic_getters_value()[source]

Verify that getters will return the actual value

test_invalid_setattr_raises()[source]

Verify that __setattr__ cannot be called on a terminal Port

test_job_calculation_get_builder_restart()[source]

Test the get_builder_restart method of CalcJobNode class

test_process_builder_attributes()[source]

Check that the builder has all the input ports of the process class as attributes

test_process_builder_set_attributes()[source]

Verify that setting attributes in builder works

test_workchain()[source]

Verify that the attributes of the TestWorkChain can be set but defaults are not there

class aiida.backends.tests.engine.test_process_builder.TestWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_process_builder'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]

Tests for the process_function decorator.

class aiida.backends.tests.engine.test_process_function.TestProcessFunction(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Note that here we use @workfunctions and @calculations, the concrete versions of the @process_function decorator, even though we are testing only the shared functionality that is captured in the @process_function decorator, relating to the transformation of the wrapped function into a FunctionProcess. The reason we do not use the @process_function decorator itself, is because it does not have a node class by default. We could create one on the fly, but then anytime inputs or outputs would be attached to it in the tests, the validate_link function would complain as the dummy node class is not recognized as a valid process node.

__module__ = 'aiida.backends.tests.engine.test_process_function'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_exit_status()[source]

A FINISHED process function has to have an exit status of 0

test_function_args()[source]

Simple process function that defines a single positional argument.

test_function_args_and_kwargs()[source]

Simple process function that defines a positional argument and keyword arguments.

test_function_args_and_kwargs_default()[source]

Simple process function that defines a positional argument and an argument with a default.

test_function_args_passing_kwargs()[source]

Cannot pass kwargs if the function does not explicitly define it accepts kwargs.

test_function_args_with_default()[source]

Simple process function that defines a single argument with a default.

test_function_defaults()[source]

Verify that a process function can define a default label and description but can be overriden.

test_function_kwargs()[source]

Simple process function that defines keyword arguments.

test_function_set_label_description()[source]

Verify that the label and description can be set for all process function variants.

test_function_varargs()[source]

Variadic arguments are not supported and should raise.

test_function_with_none_default()[source]

Simple process function that defines a keyword with None as default value.

test_hashes()[source]

Test that the hashes generated for identical process functions with identical inputs are the same.

test_hashes_different()[source]

Test that the hashes generated for identical process functions with different inputs are the different.

test_launchers()[source]

Verify that the various launchers are working.

test_normal_exception()[source]

If a process, for example a FunctionProcess, excepts, the exception should be stored in the node.

test_process_state()[source]

Test the process state for a process function.

test_return_exit_code()[source]

A process function that returns an ExitCode namedtuple should have its exit status and message set FINISHED

test_simple_workflow()[source]

Test construction of simple workflow by chaining process functions.

test_source_code_attributes()[source]

Verify function properties are properly introspected and stored in the nodes attributes and repository.

Tests for the ProcessSpec class.

class aiida.backends.tests.engine.test_process_spec.TestProcessSpec(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for the ProcessSpec class.

__module__ = 'aiida.backends.tests.engine.test_process_spec'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_dynamic_input()[source]

Test a process spec with dynamic input enabled.

test_dynamic_output()[source]

Test a process spec with dynamic output enabled.

test_exit_code()[source]

Test the definition of error codes through the ProcessSpec.

test_exit_code_invalid()[source]

Test type validation for registering new error codes.

class aiida.backends.tests.engine.test_rmq.TestProcessControl(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test AiiDA’s RabbitMQ functionalities.

TIMEOUT = 2.0
__module__ = 'aiida.backends.tests.engine.test_rmq'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_exception_process()[source]
test_kill()[source]

Test sending a kill message.

test_launch_with_inputs()[source]
test_pause()[source]

Testing sending a pause message to the process.

test_pause_play()[source]

Test sending a pause and then a play message.

test_submit_bad_input()[source]
test_submit_simple()[source]
wait_for_calc(**kwargs)[source]
wait_future(**kwargs)[source]
aiida.backends.tests.engine.test_rmq.with_timeout(*args, **kwargs)[source]

Tests for the run functions.

class aiida.backends.tests.engine.test_run.TestRun(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for the run functions.

__module__ = 'aiida.backends.tests.engine.test_run'
static test_run()[source]

Test the run function.

test_run_get_node()[source]

Test the run_get_node function.

class aiida.backends.tests.engine.test_runners.Proc(inputs=None, logger=None, runner=None, parent_pid=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.process.Process

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_runners'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
_node_class

alias of aiida.orm.nodes.process.workflow.workflow.WorkflowNode

run()[source]
class aiida.backends.tests.engine.test_runners.TestWorkchain(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_runners'
_run_loop_for(seconds)[source]
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_call_on_calculation_finish()[source]
aiida.backends.tests.engine.test_runners.the_hans_klok_comeback(loop)[source]
class aiida.backends.tests.engine.test_transport.TestTransportQueue(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for the transport queue

__module__ = 'aiida.backends.tests.engine.test_transport'
setUp(*args, **kwargs)[source]

Set up a simple authinfo and for later use

tearDown(*args, **kwargs)[source]

Hook method for deconstructing the test fixture after testing it.

test_get_transport_interleaved()[source]

Test interleaved calls to get the same transport

test_get_transport_nested()[source]

Test nesting calls to get the same transport

test_open_fail()[source]

Test that if opening fails

test_return_from_context()[source]

Test raising a Return from coroutine context

test_safe_interval()[source]

Verify that the safe interval for a given in transport is respected by the transport queue.

test_simple_request()[source]

Test a simple transport request

class aiida.backends.tests.engine.test_utils.RefObjectsStore(methodName='runTest')[source]

Bases: unittest.case.TestCase

__module__ = 'aiida.backends.tests.engine.test_utils'
test_construct()[source]

Test that construction only gets called when used

test_get_no_constructor()[source]

Test that trying to get an object that does exists and providing no means to construct it fails

test_simple()[source]

Test the reference counting works

class aiida.backends.tests.engine.test_utils.TestExponentialBackoffRetry(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for the exponential backoff retry coroutine.

__module__ = 'aiida.backends.tests.engine.test_utils'
classmethod setUpClass(*args, **kwargs)[source]

Set up a simple authinfo and for later use.

test_exponential_backoff_max_attempts_exceeded()[source]

Test that exponential backoff will finally raise if max_attempts is exceeded

test_exponential_backoff_success()[source]

Test that exponential backoff will successfully catch exceptions as long as max_attempts is not exceeded.

class aiida.backends.tests.engine.test_work_chain.ChildExposeWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
do_run()[source]
class aiida.backends.tests.engine.test_work_chain.GrandParentExposeWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
do_run()[source]
finalize()[source]
class aiida.backends.tests.engine.test_work_chain.IfTest(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
condition()[source]
classmethod define(spec)[source]
on_create(*args, **kwargs)[source]
step1()[source]
step2()[source]
class aiida.backends.tests.engine.test_work_chain.ParentExposeWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

_Process__called = True
__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
_spec = <aiida.engine.processes.workchains.workchain.WorkChainSpec object>
classmethod define(spec)[source]
finalize()[source]
start_children()[source]
class aiida.backends.tests.engine.test_work_chain.PotentialFailureWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

EXIT_MESSAGE = 'Well you did ask for it'
EXIT_STATUS = 1
OUTPUT_LABEL = 'optional_output'
OUTPUT_VALUE = 144
__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
failure()[source]
should_return_out_of_outline()[source]
success()[source]
class aiida.backends.tests.engine.test_work_chain.SerializeWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
do_test()[source]
class aiida.backends.tests.engine.test_work_chain.TestContext(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_work_chain'
test_attributes()[source]
test_dict()[source]
class aiida.backends.tests.engine.test_work_chain.TestDefaultUniqueness(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test that default inputs of exposed nodes will get unique UUIDS.

class Child(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
run()[source]
class Parent(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
do_run()[source]
__module__ = 'aiida.backends.tests.engine.test_work_chain'
test_unique_default_inputs()[source]

The default value for the Child will be constructed at import time, which will be an unstored Bool node with a given ID. When expose_inputs is called on the ProcessSpec of the Parent workchain, for the Child workchain, the ports of the Child will be deepcopied into the portnamespace of the Parent, in this case twice, into different namespaces. The port in each namespace will have a deepcopied version of the unstored Bool node. When the Parent workchain is now called without inputs, both those nodes will be stored and used as inputs, but they will have the same UUID, unless the deepcopy will have guaranteed that a new UUID is generated for unstored nodes.

class aiida.backends.tests.engine.test_work_chain.TestExitStatus(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

This class should test the various ways that one can exit from the outline flow of a WorkChain, other than it running it all the way through. Currently this can be done directly in the outline by calling the return_ construct, or from an outline step function by returning a non-zero integer or an ExitCode with a non-zero status

__module__ = 'aiida.backends.tests.engine.test_work_chain'
test_failing_workchain_through_exit_code()[source]
test_failing_workchain_through_integer()[source]
test_return_out_of_outline()[source]
test_successful_workchain_through_exit_code()[source]
test_successful_workchain_through_integer()[source]
class aiida.backends.tests.engine.test_work_chain.TestImmutableInputWorkchain(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test that inputs cannot be modified

__module__ = 'aiida.backends.tests.engine.test_work_chain'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_immutable_input()[source]

Check that from within the WorkChain self.inputs returns an AttributesFrozendict which should be immutable

test_immutable_input_groups()[source]

Check that namespaced inputs also return AttributeFrozendicts and are hence immutable

class aiida.backends.tests.engine.test_work_chain.TestSerializeWorkChain(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test workchains with serialized input / output.

__module__ = 'aiida.backends.tests.engine.test_work_chain'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_serialize()[source]

Test a simple serialization of a class to its identifier.

test_serialize_builder()[source]

Test serailization when using a builder.

class aiida.backends.tests.engine.test_work_chain.TestWorkChainAbort(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the functionality to abort a workchain

class AbortableWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
begin()[source]
check()[source]
classmethod define(spec)[source]
__module__ = 'aiida.backends.tests.engine.test_work_chain'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_simple_kill_through_process()[source]

Run the workchain for one step and then kill it by calling kill on the workchain itself. This should have the workchain end up in the KILLED state.

test_simple_run()[source]

Run the workchain which should hit the exception and therefore end up in the EXCEPTED state

class aiida.backends.tests.engine.test_work_chain.TestWorkChainAbortChildren(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the functionality to abort a workchain and verify that children are also aborted appropriately

class MainWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
check()[source]
classmethod define(spec)[source]
submit_child()[source]
class SubWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
begin()[source]

If the Main should be killed, pause the child to give the Main a chance to call kill on its children

check()[source]
classmethod define(spec)[source]
__module__ = 'aiida.backends.tests.engine.test_work_chain'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_simple_kill_through_process()[source]

Run the workchain for one step and then kill it. This should have the workchain and its children end up in the KILLED state.

test_simple_run()[source]

Run the workchain which should hit the exception and therefore end up in the EXCEPTED state

class aiida.backends.tests.engine.test_work_chain.TestWorkChainExpose(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the expose inputs / outputs functionality

__module__ = 'aiida.backends.tests.engine.test_work_chain'
test_expose()[source]
test_issue_1741_expose_inputs()[source]

Test that expose inputs works correctly when copying a stored default value

test_nested_expose(**kwargs)[source]
class aiida.backends.tests.engine.test_work_chain.TestWorkChainReturnDict(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

class PointlessWorkChain(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
classmethod define(spec)[source]
return_dict()[source]

Only return a dictionary, which should be allowed, even though it accomplishes nothing.

__module__ = 'aiida.backends.tests.engine.test_work_chain'
test_run_pointless_workchain()[source]

Running the pointless workchain should not incur any exceptions

class aiida.backends.tests.engine.test_work_chain.TestWorkchain(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.engine.test_work_chain'
_run_with_checkpoints(wf_class, inputs=None)[source]
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_checkpointing()[source]
test_context()[source]
test_exit_codes()[source]
test_if_block_persistence()[source]

This test was created to capture issue #902

test_incorrect_outline()[source]
test_malformed_outline()[source]

Test some malformed outlines

test_namespace_nondb_mapping()[source]

Regression test for a bug in _flatten_inputs

test_nondb_dynamic()[source]

Test that non-db inputs can be passed in a dynamic input namespace.

test_persisting()[source]
test_report_dbloghandler()[source]

Test whether the WorkChain, through its Process, has a logger set for which the DbLogHandler has been attached. Because if this is not the case, the ‘report’ method will not actually hit the DbLogHandler and the message will not be stored in the database

test_return()[source]
test_run()[source]
test_run_base_class()[source]

Verify that it is impossible to run, submit or instantiate a base WorkChain class.

test_same_input_node()[source]
test_str()[source]
test_to_context()[source]
test_tocontext_schedule_workchain()[source]
test_tocontext_submit_workchain_no_daemon()[source]
test_unstored_nodes_in_context()[source]
class aiida.backends.tests.engine.test_work_chain.Wf(inputs=None, logger=None, runner=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.workchains.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.engine.test_work_chain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 102
_abc_registry = <_weakrefset.WeakSet object>
_set_finished(function_name)[source]
classmethod define(spec)[source]
finished_steps = {}
isA()[source]
isB()[source]
ltN()[source]
on_create()[source]
s1()[source]
s2()[source]
s3()[source]
s4()[source]
s5()[source]
s6()[source]
aiida.backends.tests.engine.test_work_chain.run_and_check_success(process_class, **kwargs)[source]

Instantiates the process class and executes it followed by a check that it is finished successfully

Returns:instance of process
aiida.backends.tests.engine.test_work_chain.run_until_paused(proc)[source]

Set up a future that will be resolved on entering the WAITING state

aiida.backends.tests.engine.test_work_chain.run_until_waiting(proc)[source]

Set up a future that will be resolved on entering the WAITING state

Tests for the workfunction decorator and WorkFunctionNode.

class aiida.backends.tests.engine.test_workfunctions.TestWorkFunction(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for workfunctions.

__module__ = 'aiida.backends.tests.engine.test_workfunctions'
setUp()[source]

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

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

test_workfunction_caching()[source]

Verify that a workfunction cannot be cached.

test_workfunction_default_linkname()[source]

Verify that a workfunction that returns a single Data node gets a default link label.

Verify that a workfunction can only get RETURN links and no CREATE links.

test_workfunction_node_type()[source]

Verify that a workfunction gets a WorkFunctionNode as node instance.

test_workfunction_return_unstored()[source]

Verify that a workfunction will raise when an unstored node is returned.