aiida.backends.tests.engine package

Submodules

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(seconds=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_impl = <_abc_data 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_pk()[source]
test_submit_store_provenance_false()[source]

Verify that submitting with store_provenance=False raises.

test_workchain_builder_run()[source]
test_workchain_builder_run_get_node()[source]
test_workchain_builder_run_get_pk()[source]
test_workchain_run()[source]
test_workchain_run_get_node()[source]
test_workchain_run_get_pk()[source]
class aiida.backends.tests.engine.test_launch.TestLaunchersDryRun(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the launchers when performing a dry-run.

__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_launchers_dry_run()[source]

All launchers should work with dry_run=True, even submit which forwards to run.

test_launchers_dry_run_no_provenance()[source]

Test the launchers in dry_run mode with store_provenance=False.

aiida.backends.tests.engine.test_launch.add(a, b)[source]

Tests for the classes in aiida.engine.processes.calcjobs.manager.

class aiida.backends.tests.engine.test_manager.TestJobManager(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the aiida.engine.processes.calcjobs.manager.JobManager class.

__module__ = 'aiida.backends.tests.engine.test_manager'
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_get_jobs_list()[source]

Test the JobManager.get_jobs_list method.

test_request_job_info_update()[source]

Test the JobManager.request_job_info_update method.

class aiida.backends.tests.engine.test_manager.TestJobsList(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the aiida.engine.processes.calcjobs.manager.JobsList class.

__module__ = 'aiida.backends.tests.engine.test_manager'
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_get_minimum_update_interval()[source]

Test the JobsList.get_minimum_update_interval method.

test_last_updated()[source]

Test the JobsList.last_updated method.

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]

Tests for process spec ports.

class aiida.backends.tests.engine.test_ports.TestInputPort(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for the InputPort class.

__module__ = 'aiida.backends.tests.engine.test_ports'
test_with_non_db()[source]

Test the functionality of the non_db attribute upon construction and setting.

class aiida.backends.tests.engine.test_ports.TestPortNamespace(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Tests for the PortNamespace class.

__module__ = 'aiida.backends.tests.engine.test_ports'
test_lambda_default()[source]

Test that an input port can specify a lambda as a default.

test_serialize_type_check()[source]

Test that serialize will include full port namespace in exception message.

test_validate_port_name()[source]

This test will ensure that illegal port names will raise a ValueError when trying to add it.

test_with_non_db()[source]

Ports inserted to a PortNamespace should inherit the non_db attribute if not explicitly set.

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_impl = <_abc_data 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_impl = <_abc_data object>
_node_class

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

on_create()[source]

Called when a Process is created.

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_exit_codes()[source]

Test the properties to return various (sub) sets of existing exit codes.

test_input_after_stored()[source]

Verify that adding an input link after storing a ProcessNode will raise because it is illegal.

test_inputs()[source]
test_label()[source]
test_none_input()[source]
test_output_dictionary()[source]

Verify that a dictionary can be passed as an output for a namespace.

test_output_validation_error()[source]

Test that a process is marked as failed if its output namespace validation fails.

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_spec_metadata_property()[source]

Process.spec_metadata should return the metadata port namespace of its spec.

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.MappingData(data=None)[source]

Bases: collections.abc.Mapping, aiida.orm.nodes.data.data.Data

Data sub class that is also a Mapping.

__abstractmethods__ = frozenset({})
__dict__ = mappingproxy({'__module__': 'aiida.backends.tests.engine.test_process_builder', '__doc__': 'Data sub class that is also a `Mapping`.', '__init__': <function MappingData.__init__>, '__getitem__': <function MappingData.__getitem__>, '__iter__': <function MappingData.__iter__>, '__len__': <function MappingData.__len__>, '__dict__': <attribute '__dict__' of 'MappingData' objects>, '__weakref__': <attribute '__weakref__' of 'MappingData' objects>, '__abstractmethods__': frozenset(), '_abc_impl': <_abc_data object>, '_logger': <Logger aiida.backends.tests.engine.test_process_builder.MappingData (REPORT)>, '_plugin_type_string': 'aiida.backends.tests.engine.test_process_builder.MappingData.', '_query_type_string': 'aiida.backends.tests.engine.test_process_builder.'})
__getitem__(key)[source]
__init__(data=None)[source]

Initialize self. See help(type(self)) for accurate signature.

__iter__()[source]
__len__()[source]
__module__ = 'aiida.backends.tests.engine.test_process_builder'
__weakref__

list of weak references to the object (if defined)

_abc_impl = <_abc_data object>
_logger = <Logger aiida.backends.tests.engine.test_process_builder.MappingData (REPORT)>
_plugin_type_string = 'aiida.backends.tests.engine.test_process_builder.MappingData.'
_query_type_string = 'aiida.backends.tests.engine.test_process_builder.'
class aiida.backends.tests.engine.test_process_builder.TestLazyProcessNamespace(inputs=None, logger=None, runner=None, parent_pid=None, enable_persistence=True)[source]

Bases: aiida.engine.processes.process.Process

Process with basic nested namespaces to test “pruning” of empty nested namespaces from the builder.

__abstractmethods__ = frozenset({})
__module__ = 'aiida.backends.tests.engine.test_process_builder'
_abc_impl = <_abc_data object>
classmethod define(spec)[source]
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_builder_inputs()[source]

Test the ProcessBuilder._inputs method to get the inputs with and without prune set to True.

test_builder_restart_work_chain()[source]

Verify that nested namespaces imploded into flat link labels can be reconstructed into nested namespaces.

test_calc_job_node_get_builder_restart()[source]

Test the CalcJobNode.get_builder_restart method.

test_code_get_builder()[source]

Test that the Code.get_builder method 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_dynamic_setters()[source]

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

test_port_names_overlapping_mutable_mapping_methods()[source]

Check that port names take precedence over collections.MutableMapping methods.

The ProcessBuilderNamespace is a collections.MutableMapping but since the port names are made accessible as attributes, they can overlap with some of the mappings builtin methods, e.g. values(), items() etc. The port names should take precendence in this case and if one wants to access the mapping methods one needs to cast the builder to a dictionary first.

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.

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_impl = <_abc_data 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_default_label()[source]

Verify unless specified label is taken from function name.

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_out_unstored()[source]

A workfunction that returns an unstored node should raise as it indicates users tried to create data.

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_plugin_version()[source]

Test the version attributes of a process function.

test_process_state()[source]

Test the process state for a process function.

test_process_type()[source]

Test that the process type correctly contains the module and name of original decorated function.

test_properties()[source]

Test that the is_process_function and node_class attributes are set.

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(calc_node, timeout=2.0)[source]
wait_future(future, timeout=2.0)[source]
aiida.backends.tests.engine.test_rmq.with_timeout(what, timeout=5.0)[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_impl = <_abc_data 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.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.

test_is_process_function()[source]

Test the is_process_function utility.

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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data object>
condition()[source]
classmethod define(spec)[source]
on_create(*args, **kwargs)[source]

Called when a Process is created.

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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data 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_impl = <_abc_data 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()[source]
class aiida.backends.tests.engine.test_work_chain.TestWorkChainMisc(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

class IllegalSubmitWorkChain(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_impl = <_abc_data object>
classmethod define(spec)[source]
illegal_submit()[source]

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

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_impl = <_abc_data 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_global_submit_raises()[source]

Using top-level submit should raise.

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 that the call_link_label metadata input is properly used and set.

test_checkpointing()[source]
test_context()[source]
test_define_not_calling_super()[source]

A WorkChain that does not call super in define classmethod should raise.

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_out_unstored()[source]

Calling self.out on an unstored Node should raise.

It indicates that users created new data whose provenance will be lost.

test_persisting()[source]
test_process_status_sub_processes()[source]

Test that process status is set on node when waiting for sub processes.

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_impl = <_abc_data object>
_set_finished(function_name)[source]
classmethod define(spec)[source]
finished_steps = {}
isA()[source]
isB()[source]
ltN()[source]
on_create()[source]

Called when a Process is created.

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.

Verify that setting a call_link_label on a calcfunction called from a workfunction works.

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.