aiida.tools.pytest_fixtures package#

Fixtures to simplify writing unit tests for AiiDA with pytest.

Submodules#

Fixtures to create and load temporary AiiDA configuration directories and profiles.

aiida.tools.pytest_fixtures.configuration.aiida_config(tmp_path_factory, aiida_config_factory)[source]#

Return a loaded temporary AiiDA configuration directory.

This fixture is session-scoped and used automatically as soon as these fixtures are imported.

:returns Config: The loaded temporary config.

aiida.tools.pytest_fixtures.configuration.aiida_config_factory()[source]#

Return a factory to create and load a new temporary AiiDA configuration directory.

The factory is a context manager that returns a loaded aiida.manage.configuration.config.Config. It requires a path on the local file system where the configuration directory is to be created as an argument. If another configuration directory was already loaded that is automatically restored at the end of the context manager. This way, any changes made to the configuration during the context are fully temporary and automatically undone after the test.

Usage:

def test(aiida_config_factory, tmp_path_factory):
    import secrets
    with aiida_config_factory(tmp_path_factory.mktemp(secrets.token_hex(16))) as config:
        yield config

The factory has the following signature to allow further configuring the profile that is created and loaded:

Parameters:: dirpath – The path to create the configuration directory in.
Returns ~aiida.manage.configuration.config.Config~aiida.manage.configuration.config.Config:: The loaded temporary config.

aiida.tools.pytest_fixtures.configuration.aiida_config_tmp(tmp_path, aiida_config_factory)[source]#

Create and load a temporary AiiDA configuration directory.

This fixture is function-scoped and automatically restores any previously loaded config after the test.

:returns Config: The loaded temporary config.

aiida.tools.pytest_fixtures.configuration.aiida_profile(aiida_config, aiida_profile_factory)[source]#

Return a loaded temporary AiiDA profile.

This fixture is session-scoped and used automatically as soon as these fixtures are imported. The profile defines no broker and uses the core.sqlite_dos storage backend, meaning it requires no services to run.

:returns Profile: The loaded temporary profile.

aiida.tools.pytest_fixtures.configuration.aiida_profile_clean(aiida_profile)[source]#

Return a loaded temporary AiiDA profile where the data storage is cleaned before the start of the test.

This is a function-scoped version of the aiida_profile fixture.

:returns Profile: The loaded temporary profile.

aiida.tools.pytest_fixtures.configuration.aiida_profile_clean_class(aiida_profile)[source]#

Return a loaded temporary AiiDA profile where the data storage is cleaned before the start of the test.

This is a class-scoped version of the aiida_profile fixture.

Returns ~aiida.manage.configuration.profile.Profile~aiida.manage.configuration.profile.Profile:: The loaded temporary profile.

aiida.tools.pytest_fixtures.configuration.aiida_profile_factory()[source]#

Return a factory to create and load a new temporary AiiDA profile.

The factory is a context manager that returns a loaded aiida.manage.configuration.profile.Profile. It requires a aiida.manage.configuration.config.Config instance to which the profile is added. If another profile was already loaded, that is automatically restored at the end of the context manager. This way, any changes made to the profile during the context are fully temporary and automatically undone after the test. The created Profile instance dynamically has the method reset_storage added which, when called, deletes all content of the storage, recreating the default user. The daemon is also stopped if it was running.

Usage:

def test(aiida_config_tmp, aiida_profile_factory):
    with aiida_profile_factory(aiida_config_tmp) as profile:
        yield profile

The factory has the following signature to allow further configuring the profile that is created and loaded:

Parameters:

storage_backend – The storage plugin to use. Defaults to core.sqlite_dos.
storage_config – The configuration to use for the selected storage plugin.
broker_backend – The broker plugin to use. Defaults to defining no broker.
broker_config – The configuration to use for the selected broker plugin.
name – The name of the profile. Defaults to a random string.
name – The email to use for the default user. Defaults to test@localhost.

Returns ~aiida.manage.configuration.profile.Profile~aiida.manage.configuration.profile.Profile:

The loaded temporary profile.

aiida.tools.pytest_fixtures.configuration.aiida_profile_tmp(aiida_config_tmp, aiida_profile_factory)[source]#

Create and load a temporary AiiDA profile.

This fixture is function-scoped and automatically restores any previously loaded profile after the test.

:returns Profile: The loaded temporary profile.

Fixtures to interact with the daemon.

aiida.tools.pytest_fixtures.daemon.daemon_client(aiida_profile)[source]#

Return a daemon client for the configured test profile for the test session.

The daemon will be automatically stopped at the end of the test session.

Usage:

def test(daemon_client):
    from aiida.engine.daemon.client import DaemonClient
    assert isinstance(daemon_client, DaemonClient)

aiida.tools.pytest_fixtures.daemon.started_daemon_client(daemon_client)[source]#

Ensure that the daemon is running for the test profile and return the associated client.

Usage:

def test(started_daemon_client):
    assert started_daemon_client.is_daemon_running

aiida.tools.pytest_fixtures.daemon.stopped_daemon_client(daemon_client)[source]#

Ensure that the daemon is not running for the test profile and return the associated client.

Usage:

def test(stopped_daemon_client):
    assert not stopped_daemon_client.is_daemon_running

aiida.tools.pytest_fixtures.daemon.submit_and_await(started_daemon_client)[source]#

Return a factory to submit a process and wait for it to achieve the given state.

This fixture automatically loads the started_daemon_client fixture ensuring the daemon is already running, therefore it is not necessary to manually start the daemon.

Usage:

def test(submit_and_await):
    inputs = {
        ...
    }
    node = submit_and_await(SomeProcess, **inputs)

The factory has the following signature:

Parameters:

submittable – A process, a process builder or a process node. If it is a process or builder, it is submitted first before awaiting the desired state.
state – The process state to wait for, by default it waits for the submittable to be FINISHED.
timeout – The time to wait for the process to achieve the state.
kwargs – If the submittable is a process class, it is instantiated with the kwargs as inputs.

Raises:

RuntimeError – If the process fails to achieve the specified state before the timeout expires.

Returns ~aiida.orm.nodes.process.process.ProcessNode~aiida.orm.nodes.process.process.ProcessNode:

The process node.

Fixtures to temporarily add and remove entry points.

class aiida.tools.pytest_fixtures.entry_points.EntryPointManager(entry_points: EntryPoints)[source]#

Bases: object

Manager to temporarily add or remove entry points.

__dict__ = mappingproxy({'__module__': 'aiida.tools.pytest_fixtures.entry_points', '__doc__': 'Manager to temporarily add or remove entry points.', '__init__': <function EntryPointManager.__init__>, 'eps': <function EntryPointManager.eps>, 'eps_select': <function EntryPointManager.eps_select>, '_validate_entry_point': <staticmethod(<function EntryPointManager._validate_entry_point>)>, 'add': <function EntryPointManager.add>, 'remove': <function EntryPointManager.remove>, '__dict__': <attribute '__dict__' of 'EntryPointManager' objects>, '__weakref__': <attribute '__weakref__' of 'EntryPointManager' objects>, '__annotations__': {}})#

__init__(entry_points: EntryPoints)[source]#

__module__ = 'aiida.tools.pytest_fixtures.entry_points'#

__weakref__#: list of weak references to the object (if defined)

static _validate_entry_point(entry_point_string: str | None, group: str | None, name: str | None) → tuple[str, str][source]#

Validate the definition of the entry point.

Parameters:

entry_point_string – Fully qualified entry point string.
name – Entry point name.
group – Entry point group.

Returns:

The entry point group and name.

Raises:

TypeError – If entry_point_string, group or name are not a string, when defined.
ValueError – If entry_point_string is not defined, nor a group and name.
ValueError – If entry_point_string is not a complete entry point string with group and name.

add(value: type | str, entry_point_string: str | None = None, *, name: str | None = None, group: str | None = None) → None[source]#

Add an entry point.

Parameters:

value – The class or function to register as entry point. The resource needs to be importable, so it can’t be inlined. Alternatively, the fully qualified name can be passed as a string.
entry_point_string – Fully qualified entry point string.
name – Entry point name.
group – Entry point group.

Returns:

The entry point group and name.

Raises:

TypeError – If entry_point_string, group or name are not a string, when defined.
ValueError – If entry_point_string is not defined, nor a group and name.
ValueError – If entry_point_string is not a complete entry point string with group and name.

eps() → EntryPoints[source]#

eps_select(group, name=None) → EntryPoints[source]#

remove(entry_point_string: str | None = None, *, name: str | None = None, group: str | None = None) → None[source]#

Remove an entry point.

Parameters:

value – Entry point value, fully qualified import path name.
entry_point_string – Fully qualified entry point string.
name – Entry point name.
group – Entry point group.

Returns:

The entry point group and name.

Raises:

TypeError – If entry_point_string, group or name are not a string, when defined.
ValueError – If entry_point_string is not defined, nor a group and name.
ValueError – If entry_point_string is not a complete entry point string with group and name.

aiida.tools.pytest_fixtures.entry_points.entry_points(monkeypatch) → Generator[EntryPointManager, None, None][source]#

Return an instance of the EntryPointManager which allows to temporarily add or remove entry points.

This fixture monkey patches the entry point caches returned by the aiida.plugins.entry_point.eps() and aiida.plugins.entry_point.eps_select() functions to class methods of the EntryPointManager so that we can dynamically add and/or remove entry points.

Usage:

def test(entry_points):
    entry_points.add(SomeCalcJob, 'aiida.calculations:some.entry_point')
    # or, alternatively
    entry_points.add(SomeCalcJob, group='aiida.calculations', name='some.entry_point')

Fixtures that provide access to global singletons.

aiida.tools.pytest_fixtures.globals.aiida_manager() → Manager[source]#

Return the global Manager instance.

:returns Manager: The global manager instance.

Fixtures that provides ORM instances.

aiida.tools.pytest_fixtures.orm.aiida_code()[source]#

Return a factory to create a new or load an existing aiida.orm.nodes.data.code.abstract.AbstractCode.

Usage:

def test(aiida_localhost, aiida_code):
    from aiida.orm import InstalledCode
    code = aiida_code(
        'core.code.installed',
        label='test-code',
        computer=aiida_localhost,
        filepath_executable='/bin/bash'
    )
    assert isinstance(code, InstalledCode)

The factory has the following signature:

Parameters:

entry_point – Entry point of the code plugin.
label – The label of the code. Default to a randomly generated string.
kwargs – Additional keyword arguments that are passed to the code’s constructor.

Returns:

The created or loaded code instance.

aiida.tools.pytest_fixtures.orm.aiida_code_installed(aiida_code, aiida_localhost)[source]#

Return a factory to create a new or load an existing aiida.orm.nodes.data.code.installed.InstalledCode.

Usage:

def test(aiida_code_installed):
    from aiida.orm import InstalledCode
    code = aiida_code_installed()
    assert isinstance(code, InstalledCode)

The factory has the following signature:

Parameters:

label – The label of the code. Default to a randomly generated string.
default_calc_job_plugin – Optional default calcjob plugin to set.
computer – The computer to set. Defaults to localhost computer of the aiida_localhost fixture.
filepath_executable – The filepath of the executable. Defaults to /bin/bash.
use_double_quotes – Whether the executable and arguments of the code in the submission script should be escaped with single or double quotes.
with_mpi – Whether the executable should be run as an MPI program.
prepend_text – Optional bash commands that should be executed in the submission script before the executable.
append_text – Optional bash commands that should be executed in the submission script after the executable.

Returns:

The created or loaded code instance.

aiida.tools.pytest_fixtures.orm.aiida_computer(tmp_path) → t.Callable[[], 'Computer'][source]#

Return a factory to create a new or load an existing aiida.orm.computers.Computer instance.

The database is queried for an existing computer with the same label, hostname, scheduler_type and transport_type. If it exists, it means it was probably created by this fixture in a previous call and it is simply returned. Otherwise a new instance is created. Note that the computer is not explicitly configured, unless configure_kwargs are specified. By default the localhost hostname is used with the core.direct and core.local scheduler and transport plugins.

The factory has the following signature:

Parameters:

label – The computer label. If not specified, a random UUID4 is used.
hostname – The hostname of the computer. Defaults to localhost.
scheduler_type – The scheduler plugin to use. Defaults to core.direct.
transport_type – The transport plugin to use. Defaults to core.local.
minimum_job_poll_interval – The default minimum job poll interval to set. Defaults to 0.
default_mpiprocs_per_machine – The default number of MPI procs to set. Defaults to 1.
configuration_kwargs – Optional keyword arguments that, if defined, are used to configure the computer by calling aiida.orm.computers.Computer.configure().

Returns:

A stored computer instance.

aiida.tools.pytest_fixtures.orm.aiida_computer_local(aiida_computer) → t.Callable[[], Computer][source]#

Factory to return a aiida.orm.computers.Computer instance with core.local transport.

Usage:

def test(aiida_computer_ssh):
    computer = aiida_computer_ssh(label='some-label', configure=True)
    assert computer.transport_type == 'core.local'
    assert computer.is_configured

The factory has the following signature:

Parameters:

label – The computer label. If not specified, a random UUID4 is used.
configure – Boolean, if True, ensures the computer is configured, otherwise the computer is returned as is. Note that if a computer with the given label already exists and it was configured before, the computer will not be “un-“configured. If an unconfigured computer is absolutely required, make sure to first delete the existing computer or specify another label.

Returns:

A stored computer instance.

aiida.tools.pytest_fixtures.orm.aiida_computer_ssh(aiida_computer, ssh_key) → t.Callable[[], 'Computer'][source]#

Factory to return a aiida.orm.computers.Computer instance with core.ssh transport.

If configure=True, an SSH key pair is automatically added to the .ssh folder of the user, allowing an actual SSH connection to be made to the localhost.

Usage:

def test(aiida_computer_ssh):
    computer = aiida_computer_ssh(label='some-label', configure=True)
    assert computer.transport_type == 'core.ssh'
    assert computer.is_configured

The factory has the following signature:

Parameters:

label – The computer label. If not specified, a random UUID4 is used.
configure – Boolean, if True, ensures the computer is configured, otherwise the computer is returned as is. Note that if a computer with the given label already exists and it was configured before, the computer will not be “un-“configured. If an unconfigured computer is absolutely required, make sure to first delete the existing computer or specify another label.

Returns:

A stored computer instance.

aiida.tools.pytest_fixtures.orm.aiida_localhost(aiida_computer_local) → Computer[source]#

Return a aiida.orm.computers.Computer instance representing localhost with core.local transport.

Usage:

def test(aiida_localhost):
    assert aiida_localhost.transport_type == 'core.local'

Returns:: The computer.

aiida.tools.pytest_fixtures.orm.ssh_key(tmp_path_factory) → Generator[Path, None, None][source]#

Generate a temporary SSH key pair for the test session and return the filepath of the private key.

The filepath of the public key is the same as the private key, but it adds the .pub file extension.

Returns:: The filepath of the generated private key.

Fixtures providing resources for storage plugins.

aiida.tools.pytest_fixtures.storage.config_psql_dos(tmp_path_factory: pytest.TempPathFactory, postgres_cluster: PGTest) → t.Callable[[str | None, str | None, str | None], dict[str, t.Any]][source]#

Return a profile configuration for the PsqlDosBackend.

The factory has the following signature to allow further configuring the database that is created:

Parameters:

database_name – Name of the database to be created.
database_username – Username to use for authentication.
database_password – Password to use for authentication.

Returns:

The dictionary with the storage configuration for the core.psql_dos storage plugin.

aiida.tools.pytest_fixtures.storage.postgres_cluster()[source]#

Create a temporary and isolated PostgreSQL cluster using pgtest and cleanup after the yield.

Parameters:

database_name – Name of the database.
database_username – Username to use for authentication.
database_password – Password to use for authentication.

Returns:

Dictionary with parameters to connect to the PostgreSQL cluster.