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 testingConsumer
classes.-
create_message
(message, properties=None, exchange='rejected', routing_key='test')[source]¶ Create a message instance for use with the consumer in testing.
Parameters: Return type:
-
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: Raises: Raises: Return type:
-
published_messages
¶ Return a list of
PublishedMessage
that are extracted from all calls tobasic_publish()
that are invoked during the test. The properties attribute is thepika.spec.BasicProperties
instance that was created during publishing.New in version 3.18.9.
Returns: list([ PublishedMessage
])
-
-
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.