tests.test_request_task

Helpful documentation on how to setup celery test fixtures:

https://medium.com/@scythargon/how-to-use-celery-pytest-fixtures-for-celery-integration-testing-6d61c91775d9.

TL;DR : - Add a session fixture with an in-memory celery config. - Use the celery_session_app and celery_session_worker fixtures for the tests. - Don’t forget to use the shared_task decorator instead of the usual app.task because shared_task uses

a proxy allowing the task to be bound to any app (including the celery_session_app fixture).

Classes

UnreliableRequestTask	Celery base task that should be used to handle API requests.
TestRequestTask	A class whose instances are single test cases.

Functions

celery_config()
unreliable_request_task(→ int)
abort_sum_task(→ int)
sum_task(→ int)

Module Contents

tests.test_request_task.celery_config()

class tests.test_request_task.UnreliableRequestTask

Bases: cowbird.request_task.RequestTask, abc.ABC

Celery base task that should be used to handle API requests.

Using this class will set the following Task configuration :

• auto-retry for every RequestException

• backoff and jitter strategy

There is also an abort_chain function to stop the chain of requests in case of an unrecoverable event

To use this class simply decorate your asynchronous function like this:

from celery import shared_task

@shared_task(bind=True, base=RequestTask)
def function_name(self, any, wanted, parameters):
    pass  # function operations

Parameter bind=True will provide the self argument to the function which is the celery Task (not required).

Parameter base=RequestTask will instantiate a RequestTask rather than a base celery Task as the self object.

retry_jitter = False

test_max_retries = 3

invoke_time = []

__call__(*args, **kwargs)

tests.test_request_task.unreliable_request_task(self: cowbird.request_task.RequestTask, param1: int, param2: int) → int

tests.test_request_task.abort_sum_task(self: cowbird.request_task.RequestTask, param1: int, param2: int) → int

tests.test_request_task.sum_task(param1: int, param2: int) → int

class tests.test_request_task.TestRequestTask(methodName='runTest')

Bases: unittest.TestCase

A class whose instances are single test cases.

By default, the test code itself should be placed in a method named ‘runTest’.

If the fixture may be used for many test cases, create as many test methods as are needed. When instantiating such a TestCase subclass, specify in the constructor arguments the name of the test method that the instance is to execute.

Test authors should subclass TestCase for their own tests. Construction and deconstruction of the test’s environment (‘fixture’) can be implemented by overriding the ‘setUp’ and ‘tearDown’ methods respectively.

If it is necessary to override the __init__ method, the base class __init__ method must always be called. It is important that subclasses should not change the signature of their __init__ method, since instances of the classes are instantiated automatically by parts of the framework in order to be run.

When subclassing TestCase, you can set these attributes: * failureException: determines which exception will be raised when

the instance’s assertion methods fail; test methods raising this exception will be deemed to have ‘failed’ rather than ‘errored’.

• longMessage: determines whether long messages (including repr of

  objects used in assert methods) will be printed on failure in addition to any explicit message passed.

• maxDiff: sets the maximum length of a diff in failure messages

  by assert methods using difflib. It is looked up as an instance attribute so can be configured by individual tests if required.

Create an instance of the class that will use the named test method when executed. Raises a ValueError if the instance does not have a method with the specified name.

classmethod setUpClass()

Hook method for setting up class fixture before running tests in the class.

classmethod tearDownClass()

Hook method for deconstructing the class fixture after running all tests in the class.

test_chain()

test_retrying()

test_aborting()

test_geoserver_user_created(create_workspace_mock, create_datastore_mock, create_datastore_dir_mock)

test_geoserver_workspace_datastore_created(create_workspace_request_mock, create_datastore_request_mock, configure_datastore_request_mock, _create_datastore_dir_mock)

test_geoserver_user_deleted(remove_workspace_mock)

test_geoserver_workspace_removed(remove_workspace_request_mock)

test_geoserver_file_creation(validate_shapefile_mock, publish_shapefile_request_mock)

test_geoserver_file_removal(remove_shapefile_request_mock)