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()[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 theConsumerBase
fixture, this class has anon_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)[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 certainlimit
. Iflimit
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.default_exchange()[source]¶
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)[source]¶
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)[source]¶
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)[source]¶
Create users.
Scope: function
This fixture creates an user with a default UUID
00000000-0000-0000-0000-000000000000
,email
info@reana.io andaccess_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.in_memory_queue_connection()[source]¶
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.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, sample_workflow_workspace, tmp_shared_volume_path)[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, sample_workflow_workspace, tmp_shared_volume_path)[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()[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.snakemake_workflow_spec_loaded()[source]¶
Scatter-gather Snakemake workflow.
Scope: function
This fixture provides the internal representation of a basic scatter-gather
snakemake
workflow loaded as dictionary.
Fixture temporary file system database.
Scope: module
This fixture offers a temporary shared file system using
tmpdir_factory.mktemp
namedreana
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.yadage_workflow_spec_loaded()[source]¶
Nested Yadage workflow.
Scope: function
This fixture provides a nested
yadage
workflow spec loaded as 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 withPyYAML
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 fordb
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 thepath
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
andsample_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:
Search for already reported problems.
Check if the issue has been fixed or is still reproducible on the latest
master
branch.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.