2. API Docs

2.1. Fixtures

Pytest fixtures for REANA.

pytest_reana.fixtures.ConsumerBase()[source]

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

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, db_engine, session)[source]

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

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

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

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

CWL workflow without name.

Scope: function

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

pytest_reana.fixtures.db_engine(base_app)[source]

Create a SQL Alchemy DB engine.

Scope: module

This fixture offers a SQLAlchemy database engine and it expects a fixture called base_app which should be a configured Flask application including a SQLALCHEMY_DATABASE_URI configuration variable. When finished it will delete the database.

pytest_reana.fixtures.default_exchange()[source]

Return a default kombu.Exchange created from configuration.

Scope: function

This fixture offers a default kombu.Exchange.

pytest_reana.fixtures.default_in_memory_producer(in_memory_queue_connection, default_exchange)[source]

Rerturn a kombu.Producer connected to in memory queue.

Scope: function

This fixture offers a default 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)[source]

Return a default kombu.Queue created from configuration.

Scope: function

This fixture offers a default kombu.Queue.

pytest_reana.fixtures.default_user(app, session)[source]

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

Empty user secrets dictionary.

pytest_reana.fixtures.in_memory_queue_connection()[source]

In memory message queue.

Scope: session

This fixture offers an in memory 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.no_db_user()[source]

Mock user created without using db.

pytest_reana.fixtures.sample_condition_for_requeueing_workflows()[source]

Sample always false condition.

pytest_reana.fixtures.sample_condition_for_starting_queued_workflows()[source]

Sample always true condition.

pytest_reana.fixtures.sample_serial_workflow_in_db(app, default_user, session, serial_workflow)[source]

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

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

Create a sample workflow in the database.

Scope: function

Adds a sample yadage workflow in the DB.

pytest_reana.fixtures.serial_workflow()[source]

Create a serial workflow.

Scope: function

This fixture provides a serial workflow.

pytest_reana.fixtures.session(db_engine)[source]

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.tmp_shared_volume_path(tmpdir_factory)[source]

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

Test user secrets dictionary.

pytest_reana.fixtures.yadage_workflow_with_name()[source]

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

Yadage workflow without name.

Scope: function

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

2.2. Plugin

Pytest plugin for REANA.