Testing Support

The rejected.testing.AsyncTestCase provides a based class for the easy creation of tests for your consumers. The test cases exposes multiple methods to make it easy to setup a consumer and process messages. It is build on top of tornado.testing.AsyncTestCase which extends unittest.TestCase.

To get started, override the rejected.testing.AsyncTestCase.get_consumer() method.

Next, the rejected.testing.AsyncTestCase.get_settings() method can be overridden to define the settings that are passed into the consumer.

Finally, to invoke your Consumer as if it were receiving a message, the process_message() method should be invoked.

Note

Tests are asynchronous, so each test should be decorated with gen_test().

Example

The following example expects that when the message is processed by the consumer, the consumer will raise a MessageException.

from rejected import consumer, testing

import my_package


class ConsumerTestCase(testing.AsyncTestCase):

    def get_consumer(self):
        return my_package.Consumer

    def get_settings(self):
        return {'remote_url': 'http://foo'}

    @testing.gen_test
    def test_consumer_raises_message_exception(self):
        with self.assertRaises(consumer.MessageException):
            yield self.process_message({'foo': 'bar'})
class rejected.testing.AsyncTestCase(methodName: str = 'runTest')[source]

tornado.testing.AsyncTestCase subclass for testing Consumer classes.

create_message(message, properties=None, exchange='rejected', routing_key='test')[source]

Create a message instance for use with the consumer in testing.

Parameters:
  • message (any) – the body of the message to create
  • properties (dict) – AMQP message properties
  • exchange (str) – The exchange the message should appear to be from
  • routing_key (str) – The message’s routing key
Return type:

rejected.data.Message

get_consumer()[source]

Override to return the consumer class for testing.

Return type:rejected.consumer.Consumer
get_settings()[source]

Override this method to provide settings to the consumer during construction. These settings should be from the config stanza of the Consumer configuration.

Return type:dict
measurement

Return the rejected.data.Measurement for the currently assigned measurement object to the consumer.

Return type:rejected.data.Measurement
process_message(message_body=None, content_type='application/json', message_type=None, properties=None, exchange='rejected', routing_key='routing-key')[source]

Process a message as if it were being delivered by RabbitMQ. When invoked, an AMQP message will be locally created and passed into the consumer. With using the default values for the method, if you pass in a JSON serializable object, the message body will automatically be JSON serialized.

If an exception is not raised, a Measurement instance is returned that will contain all of the measurements collected during the processing of the message.

Example:

class ConsumerTestCase(testing.AsyncTestCase):

    @testing.gen_test
    def test_consumer_raises_message_exception(self):
        with self.assertRaises(consumer.MessageException):
            result = yield self.process_message({'foo': 'bar'})

Note

This method is a co-routine and must be yielded to ensure that your tests are functioning properly.

Parameters:
  • message_body (any) – the body of the message to create
  • content_type (str) – The mime type
  • message_type (str) – identifies the type of message to create
  • properties (dict) – AMQP message properties
  • exchange (str) – The exchange the message should appear to be from
  • routing_key (str) – The message’s routing key
Raises:

rejected.consumer.ConsumerException

Raises:

rejected.consumer.MessageException

Raises:

rejected.consumer.ProcessingException

Return type:

rejected.data.Measurement

published_messages

Return a list of PublishedMessage that are extracted from all calls to basic_publish() that are invoked during the test. The properties attribute is the pika.spec.BasicProperties instance that was created during publishing.

New in version 3.18.9.

Returns:list([PublishedMessage])
setUp()[source]

Hook method for setting up the test fixture before exercising it.

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

class rejected.testing.PublishedMessage(exchange, routing_key, properties, body)[source]

Contains information about messages published during a test when using rejected.testing.AsyncTestCase.

Parameters:
  • exchange (str) – The exchange the message was published to
  • routing_key (str) – The routing key the message was published with
  • properties (pika.spec.BasicProperties) – AMQP message properties
  • body (bytes) – AMQP message body

New in version 3.18.9.

rejected.testing.gen_test(func: Optional[Callable[[...], Union[collections.abc.Generator, Coroutine]]] = None, timeout: Optional[float] = None) → Union[Callable[[...], None], Callable[[Callable[[...], Union[collections.abc.Generator, Coroutine]]], Callable[[...], None]]][source]

Testing equivalent of tornado.gen.coroutine(), to be applied to test methods.