aiida.backends.tests.work package

Submodules

class aiida.backends.tests.work.class_loader.TestJobProcess(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.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.work.common.BadOutput[source]

Bases: aiida.work.process.Process

A Process that emits an output that isn’t part of the spec raising an exception.

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.common'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_run()[source]
classmethod define(spec)[source]
class aiida.backends.tests.work.common.DummyProcess[source]

Bases: aiida.work.process.Process

A Process that does nothing when it runs.

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.common'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_run()[source]
classmethod define(spec)[source]
class aiida.backends.tests.work.daemon.FailCreateFromSavedStateProcess[source]

Bases: aiida.work.test_utils.DummyProcess

This class emulates a failure that occurs when loading the process from a saved state.

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.daemon'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
on_create(pid, inputs, saved_instance_state)[source]

Called when the process is created. If a checkpoint is supplied the process should reinstate its state at the time the checkpoint was taken and if the checkpoint has a wait_on the process will continue from the corresponding callback function.

Parameters:inputs – The inputs the process should run using.
class aiida.backends.tests.work.daemon.ProcessEventsTester[source]

Bases: aiida.work.process.Process

EVENTS = ['create', 'run', 'continue_', 'finish', 'emitted', 'stop', 'destroy']
__abstractmethods__ = frozenset([])
__init__()[source]

x.__init__(…) initializes x; see help(type(x)) for signature

__module__ = 'aiida.backends.tests.work.daemon'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_on_output_emitted(output_port, value, dynamic)[source]

The process has emitted a value on the given output port.

Parameters:
  • output_port – The output port name the value was emitted on
  • value – The value emitted
  • dynamic – Was the output port a dynamic one (i.e. not known beforehand?)
_run()[source]
classmethod define(spec)[source]
finish(wait_on)[source]
on_continue(wait_on)[source]
on_create(pid, inputs, saved_instance_state)[source]

Called when the process is created. If a checkpoint is supplied the process should reinstate its state at the time the checkpoint was taken and if the checkpoint has a wait_on the process will continue from the corresponding callback function.

Parameters:inputs – The inputs the process should run using.
on_destroy()[source]

Called when a Process enters the DESTROYED state which should be the final process state and so we seal the calculation node

on_finish()[source]

Called when a Process enters the FINISHED state at which point we set the corresponding attribute of the workcalculation node

on_run()[source]

Called when the process is about to run some code either for the first time (in which case an on_start message would have been received) or after something it was waiting on has finished (in which case an on_continue message would have been received).

Any class overriding this method should make sure to call the super method, usually at the end of the function.

on_stop()[source]
on_wait(wait_on)[source]
class aiida.backends.tests.work.daemon.TestDaemon(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.daemon'
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_create_fail()[source]
test_multiple_processes()[source]
test_submit()[source]
test_tick()[source]
class aiida.backends.tests.work.persistence.TestProcess(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.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.work.process.ProcessStackTest[source]

Bases: aiida.work.process.Process

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.process'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_run()[source]
on_create(pid, inputs, saved_instance_state)[source]

Called when the process is created. If a checkpoint is supplied the process should reinstate its state at the time the checkpoint was taken and if the checkpoint has a wait_on the process will continue from the corresponding callback function.

Parameters:inputs – The inputs the process should run using.
on_stop()[source]
class aiida.backends.tests.work.process.TestFunctionProcess(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.process'
test_fixed_inputs()[source]
test_kwargs()[source]
class aiida.backends.tests.work.process.TestProcess(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.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_calculation_input()[source]
test_description()[source]
test_inputs()[source]
test_inputs_template()[source]
test_label()[source]
test_none_input()[source]
test_pid_is_uuid()[source]
test_process_stack()[source]
test_seal()[source]
class aiida.backends.tests.work.processSpec.TestProcessSpec(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.processSpec'
_test_template(template)[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_dynamic_input()[source]
test_dynamic_output()[source]
test_get_inputs_template()[source]
class aiida.backends.tests.work.process_registry.TestProcessRegistry(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.process_registry'
class aiida.backends.tests.work.run.TestRun(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.run'
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_queue_up()[source]
class aiida.backends.tests.work.tickingEngine.DummyProcess[source]

Bases: aiida.work.process.Process

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.tickingEngine'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_run(a)[source]
classmethod define(spec)[source]
class aiida.backends.tests.work.tickingEngine.TestTickingEngine(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.tickingEngine'
_keep_ticking()[source]
_tick_till_finished()[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_cancel()[source]
test_get_process()[source]
test_submit()[source]
class aiida.backends.tests.work.util.StackTester[source]

Bases: aiida.work.process.Process

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.util'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_run()[source]
class aiida.backends.tests.work.util.TestProcessRegistry(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

These these check that the registry is giving out the right pid which when using storage is equal to the node pk.

__module__ = 'aiida.backends.tests.work.util'
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_process_pid_and_calc()[source]
test_wf_pid_and_calc()[source]
aiida.backends.tests.work.util.nested_tester(*args, **kwargs)[source]
aiida.backends.tests.work.util.registry_tester(*args, **kwargs)[source]
class aiida.backends.tests.work.workChain.IfTest[source]

Bases: aiida.work.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.workChain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
condition()[source]
classmethod define(spec)[source]
on_create(pid, inputs, saved_state)[source]

Called when the process is created. If a checkpoint is supplied the process should reinstate its state at the time the checkpoint was taken and if the checkpoint has a wait_on the process will continue from the corresponding callback function.

Parameters:inputs – The inputs the process should run using.
step1()[source]
step2()[source]
class aiida.backends.tests.work.workChain.TestContext(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.workChain'
test_attributes()[source]
test_dict()[source]
class aiida.backends.tests.work.workChain.TestHelpers(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the helper functions/classes used by workchains

__module__ = 'aiida.backends.tests.work.workChain'
test_get_proc_outputs()[source]
class aiida.backends.tests.work.workChain.TestWorkChainAbort(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

Test the functionality to abort a workchain

class AbortableWorkChain[source]

Bases: aiida.work.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.workChain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
check()[source]
classmethod define(spec)[source]
start()[source]
__module__ = 'aiida.backends.tests.work.workChain'
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_node()[source]

Run the workchain for one step and then kill it by calling kill on the underlying WorkCalculation node. This should have the workchain end up in the ABORTED state.

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 ABORTED state.

test_simple_run()[source]

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

class aiida.backends.tests.work.workChain.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[source]

Bases: aiida.work.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.workChain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
check()[source]
classmethod define(spec)[source]
start()[source]
class SubWorkChain[source]

Bases: aiida.work.workchain.WorkChain

__abstractmethods__ = frozenset([])
__module__ = 'aiida.backends.tests.work.workChain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
check()[source]
classmethod define(spec)[source]
start()[source]
__module__ = 'aiida.backends.tests.work.workChain'
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_node()[source]

Run the workchain for one step and then kill it by calling kill on the underlying WorkCalculation node. This should have the workchain end up in the ABORTED state.

test_simple_run()[source]

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

class aiida.backends.tests.work.workChain.TestWorkchain(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.workChain'
_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_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_return()[source]
test_run()[source]
test_same_input_node()[source]
test_str()[source]
class aiida.backends.tests.work.workChain.TestWorkchainWithOldWorkflows(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.workChain'
test_call_old_wf()[source]
test_old_wf_results()[source]
class aiida.backends.tests.work.workChain.Wf[source]

Bases: aiida.work.workchain.WorkChain

__abstractmethods__ = frozenset([])
__init__()[source]

x.__init__(…) initializes x; see help(type(x)) for signature

__module__ = 'aiida.backends.tests.work.workChain'
_abc_cache = <_weakrefset.WeakSet object>
_abc_negative_cache = <_weakrefset.WeakSet object>
_abc_negative_cache_version = 95
_abc_registry = <_weakrefset.WeakSet object>
_set_finished(function_name)[source]
classmethod define(spec)[source]
finished_steps = {}
isA()[source]
isB()[source]
ltN()[source]
s1()[source]
s2()[source]
s3()[source]
s4()[source]
s5()[source]
s6()[source]
class aiida.backends.tests.work.workfunction.TestWf(methodName='runTest')[source]

Bases: aiida.backends.testbase.AiidaTestCase

__module__ = 'aiida.backends.tests.work.workfunction'
_check_hash_consistent(pid)[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_blocking()[source]
test_hashes()[source]
test_hashes_different()[source]
test_run()[source]
aiida.backends.tests.work.workfunction.return_input(*args, **kwargs)[source]
aiida.backends.tests.work.workfunction.simple_wf(*args, **kwargs)[source]