pytest-REANA

pytest-REANA is a component of the REANA reusable and reproducible research data analysis platform. It provides pytest fixtures and test utilities.

Features

• pytest fixtures

• mocking OpenAPI client with request format validation

• mocking Advanced Message Queuing Protocol consumers and producers

• mocking Celery tasks

• mocking file system workspace access

• mocking database access

• includes sample workspace directory

Usage

The detailed information on how to install and use REANA can be found in docs.reana.io.

API

Fixtures

Pytest fixtures for REANA.

pytest_reana.fixtures.ConsumerBase()

Return a class implementing a BaseConsumer.

Scope: function

This fixture offers a class which implements the reana_commons.BaseConsumer. It will just acknowledge the received messages.

pytest_reana.fixtures.ConsumerBaseOnMessageMock(ConsumerBase)

Return a BaseConsumer class with on_message mocked.

Scope: function

This fixture offers a class which implements the reana_commons.BaseConsumer. Additionally to the ConsumerBase fixture, this class has an on_message mocked method so actions like the following can be performed.

def test_msg(ConsumerBaseOnMessageMock)
    consumer = ConsumerBaseOnMessageMock()
    # 1 message is published with message {'some': 'message'}
    expected_body = {'some': 'message'}
    consumer.on_message.assert_called_once_with(
        expected, ANY)

pytest_reana.fixtures.app(base_app)

Flask application fixture.

Scope: function

This fixture offers a Flask application with already a database connection and all the models created. When finished it will delete all models.

def create_ninja_turtle()
    with app.test_client() as client:
        somedata = 'ninja turtle'
        res = client.post(url_for('api.create_object'),
                          content_type='application/json',
                          data=json.dumps(somedata))

        assert res.status_code == 200

pytest_reana.fixtures.consume_queue()

Provide a callable to consume a queue.

Scope: function

This fixture offers a function which given a kombu.Consumer will consume the messages in the queue until a certain limit. If limit is not specified it will consume uninterruptedly.

def test_consume_1_msg(ConsumerBase, consume_queue)
    consumer = ConsumerBase()
    # some message is published in the queue
    # and we want to consume only one.
    consume_queue(consumer, limit=1)

pytest_reana.fixtures.corev1_api_client_with_user_secrets(no_db_user)

Kubernetes CoreV1 api client with user secrets in K8s secret store.

Scope: function

Adds the CoreV1APIClient with example user secrets.

pytest_reana.fixtures.cwl_workflow_with_name()

CWL workflow with name.

Scope: function

This fixture provides a CWL workflow with a name as a dictionary.

pytest_reana.fixtures.cwl_workflow_without_name()

CWL workflow without name.

Scope: function

This fixture provides a CWL workflow without a name as a dictionary.

pytest_reana.fixtures.default_exchange()

Return a default class:kombu.Exchange created from configuration.

Scope: function

This fixture offers a default class:kombu.Exchange.

pytest_reana.fixtures.default_in_memory_producer(in_memory_queue_connection, default_exchange)

Rerturn a class:kombu.Producer connected to in memory queue.

Scope: function

This fixture offers a default class:kombu.Producer instantiated using the in_memory_queue_connection.

def test_publish_hello(default_in_memory_producer, default_queue):
    msg = {'hello': 'ninja turtle'}
    default_in_memory_producer.publish(msg,
                                       declare=[default_queue])

pytest_reana.fixtures.default_queue(default_exchange)

Return a default class:kombu.Queue created from configuration.

Scope: function

This fixture offers a default class:kombu.Queue.

pytest_reana.fixtures.default_user(app, session)

Create users.

Scope: function

This fixture creates an user with a default UUID 00000000-0000-0000-0000-000000000000, email info@reana.io and access_token secretkey and returns it.

def test_default_user_exists(default)
    with app.test_client() as client:
        res = client.post(url_for('api.get_users'),
                          query_string={"user": default_user.id_})

        assert res.status_code == 200

pytest_reana.fixtures.empty_user_secrets()

Empty user secrets dictionary.

pytest_reana.fixtures.in_memory_queue_connection()

In memory message queue.

Scope: session

This fixture offers an in memory class:kombu.Connection scoped to the testing session.

def test_something(ConsumerBase, in_memory_queue_connection):
    consumer = ConsumerBase(connection=in_memory_queue_connection)
    # Now you have a consumer connected to an in memory queue

pytest_reana.fixtures.kerberos_user_secrets()

User secrets needed by Kerberos.

pytest_reana.fixtures.no_db_user()

Mock user created without using db.

pytest_reana.fixtures.sample_condition_for_requeueing_workflows()

Sample always false condition.

pytest_reana.fixtures.sample_condition_for_starting_queued_workflows()

Sample always true condition.

pytest_reana.fixtures.sample_serial_workflow_in_db(app, default_user, session, serial_workflow, sample_workflow_workspace, tmp_shared_volume_path)

Create a sample workflow in the database.

Scope: function

Adds a sample serial workflow in the DB.

pytest_reana.fixtures.sample_workflow_workspace(tmp_shared_volume_path)

Return the directory path of a sample workspace.

Scope: module

Creates a sample workspace in the shared volume path. Copies contents from the tests/test_workspace directory.

pytest_reana.fixtures.sample_yadage_workflow_in_db(app, default_user, session, yadage_workflow_with_name, sample_workflow_workspace, tmp_shared_volume_path)

Create a sample workflow in the database.

Scope: function

Adds a sample yadage workflow in the DB.

pytest_reana.fixtures.serial_workflow()

Create a serial workflow.

Scope: function

This fixture provides a serial workflow.

pytest_reana.fixtures.session()

Create a SQL Alchemy session.

Scope: function

This fixture offers a SQLAlchemy session which has been created from the db_engine fixture.

from reana_db.models import Workflow

def test_create_workflow(session):
    workflow = Workflow(...)
    session.add(workflow)
    session.commit()

pytest_reana.fixtures.snakemake_workflow_spec_loaded()

Scatter-gather Snakemake workflow.

Scope: function

This fixture provides the internal representation of a basic scatter-gather snakemake workflow loaded as dictionary.

pytest_reana.fixtures.tmp_shared_volume_path(tmpdir_factory)

Fixture temporary file system database.

Scope: module

This fixture offers a temporary shared file system using tmpdir_factory.mktemp named reana and when scoped is finished it will be deleted.

import os

def test_dir_exists(tmp_shared_volume_path):
    path = os.path.join(
        tmp_shared_volume_path, 'directory_path')
    assert os.path.exists(path)

pytest_reana.fixtures.user_secrets()

Test user secrets dictionary.

pytest_reana.fixtures.yadage_workflow_spec_loaded()

Nested Yadage workflow.

Scope: function

This fixture provides a nested yadage workflow spec loaded as dictionary.

pytest_reana.fixtures.yadage_workflow_with_name()

Yadage workflow with name.

Scope: function

This fixture provides a yadage workflow with a name as a dictionary.

pytest_reana.fixtures.yadage_workflow_without_name()

Yadage workflow without name.

Scope: function

This fixture provides a yadage workflow without name as a dictionary.

Plugin

Pytest plugin for REANA.

Changelog

0.9.2 (2023-11-30)

• Changes CI to use the stable release of Python 3.12.

0.9.1 (2023-09-26)

• Adds support for Python 3.12.

• Changes apispec dependency version in order to be compatible with PyYAML v6.

• Fixes container image fixtures to be Podman-compatible.

• Fixes Kombu documentation linking.

0.9.0 (2022-12-13)

• Adds fixture providing example of user secrets needed for Kerberos tests.

• Adds support for Python 3.11.

• Fixes location of Celery docs for ReadTheDocs pages.

• Removes hard-dependency on black code formatter version.

0.8.1 (2022-01-05)

• Adds support for Python 3.10.

0.8.0 (2021-11-22)

• Adds nested Yadage workflow specification fixture.

• Adds empty workflow workspaces for sample workflows by default.

• Adds internal representation of a scatter-gather Snakemake workflow fixture.

• Changes tmp_shared_volume_path fixture to be configurable through environment variable.

• Changes fixtures to run with the full workspace path stored in the database.

• Removes support for Python 2.

0.7.2 (2021-07-02)

• Changes internal dependencies to remove click.

0.7.1 (2021-03-17)

• Adds support for Python 3.9.

• Fixes minor code warnings.

• Fixes installation by upgrading REANA-DB version.

0.7.0 (2020-10-20)

• Adds new __reana database schema for db fixture.

• Fixes a problem related to duplicated database session.

• Changes code formatting to respect black coding style.

• Changes documentation to single-page layout.

0.6.0 (2019-12-19)

• Adds fixtures for secrets store.

• Centralises test requirements.

• Adds Python 3.8 support.

0.5.0 (2019-04-16)

• Makes workspace path configurable for the sample_workflow_workspace fixture through the path parameter.

• Adds sample_serial_workflow_in_db fixture.

• Exposes previously hidden sample_yadage_workflow_in_db fixture.

• Adds missing database session close in session fixture.

• Adds helpers to represent starting and requeueing job conditions, sample_condition_for_starting_queued_workflows and sample_condition_for_requeueing_workflows.

0.4.1 (2018-11-06)

• Adds directory including sample workspace data.

0.4.0 (2018-11-06)

• Initial public release.

Contributing

Bug reports, issues, feature requests, and other contributions are welcome. If you find a demonstrable problem that is caused by the REANA code, please:

1. Search for already reported problems.

2. Check if the issue has been fixed or is still reproducible on the latest master branch.

3. Create an issue, ideally with a test case.

If you create a pull request fixing a bug or implementing a feature, you can run the tests to ensure that everything is operating correctly:

$ ./run-tests.sh

Each pull request should preserve or increase code coverage.

License

MIT License

Copyright (C) 2018, 2019, 2020, 2021, 2022, 2023, 2024 CERN.

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

In applying this license, CERN does not waive the privileges and immunities granted to it by virtue of its status as an Intergovernmental Organization or submit itself to any jurisdiction.

Authors

The list of contributors in alphabetical order:

• Audrius Mecionis

• Camila Diaz

• Diego Rodriguez

• Dinos Kousidis

• Giuseppe Steduto

• Jan Okraska

• Marco Donadoni

• Marco Vidal

• Rokas Maciulaitis

• Tibor Simko

• Vladyslav Moisieienkov