Test helper API reference¶

Basic test helpers¶

class vumi.tests.helpers.VumiTestCase¶

Base test case class for all things vumi-related.

This is a subclass of twisted.trial.unittest.TestCase with a small number of additional features:

It implements IHelperEnabledTestCase to make using helpers easier. (See add_helper().)
timeout is set to a default value of 5 and can be overridden by setting the VUMI_TEST_TIMEOUT environment variable. (Longer timeouts are more reliable for continuous integration builds, shorter ones are less painful for local development.)
add_cleanup() provides an alternative mechanism for specifying cleanup in the same place as the creation of thing that needs to be cleaned up.

Note

While this class does not have a setUp() method (thus avoiding the need for subclasses to call it), it does have a tearDown() method. add_cleanup() should be used in subclasses instead of overriding tearDown().

add_cleanup(func, *args, **kw)¶

Register a cleanup function to be called at teardown time.

Parameters:	func (callable) – The callable object to call at cleanup time. This callable may return a `Deferred`, in which case cleanup will continue after it fires. args – Passed to `func` when it is called. *kw – Passed to `func` when it is called.

Note

This method should be use in place of the inherited addCleanup() method, because the latter doesn’t apply timeouts to cleanup functions.

add_helper(helper_object, *args, **kw)¶

Perform setup and register cleanup for the given helper object.

Parameters:	helper_object – Helper object to add. `helper_object` must provide the `IHelper` interface. args – Passed to `helper_object.setup()` when it is called. *kw – Passed to `helper_object.setup()` when it is called.
Returns:	Either `helper_object` or a `Deferred` that fires with it.

If helper_object.setup() returns a Deferred, this method also returns a Deferred.

Example usage assuming @inlineCallbacks:

>>> @inlineCallbacks
... def test_foo(self):
...     msg_helper = yield self.add_helper(MessageHelper())
...     msg_helper.make_inbound("foo")

Example usage assuming non-async setup:

>>> def test_bar(self):
...     msg_helper = self.add_helper(MessageHelper())
...     msg_helper.make_inbound("bar")

tearDown(*args, **kwargs)¶: Run any cleanup functions registered with add_cleanup().

vumi.tests.helpers.DEFAULT¶: This constant is a placeholder value for parameter defaults. We can’t just use None because we may want to override a non-None default with an explicit None value.

interface vumi.tests.helpers.IHelper¶

Interface for test helpers.

This specifies a standard setup and cleanup mechanism used by test cases that implement the IHelperEnabledTestCase interface.

There are no interface restrictions on the constructor of a helper.

setup(*args, **kwargs)¶

Perform potentially async helper setup.

This may return a deferred for async setup or block for sync setup. All helpers must implement this even if it does nothing.

If the setup is optional but commonly used, this method can take flags to perform or suppress all or part of it as required.

cleanup()¶

Clean up any resources created by this helper.

This may return a deferred for async cleanup or block for sync cleanup. All helpers must implement this even if it does nothing.

interface vumi.tests.helpers.IHelperEnabledTestCase¶

Interface for test cases that use helpers.

This specifies a standard mechanism for managing setup and cleanup of helper classes that implement the IHelper interface.

add_helper(helper_object, *args, **kwargs)¶

Register cleanup and perform setup for a helper object.

This should call helper_object.setup(*args, **kwargs) and self.add_cleanup(helper_object.cleanup) or an equivalent.

Returns the helper_object passed in or a Deferred if setup is async.

vumi.tests.helpers.proxyable(func)¶

Mark a method as being suitable for automatic proxy generation.

See generate_proxies() for usage.

vumi.tests.helpers.generate_proxies(target, source)¶

Generate proxies on target for proxyable methods on source.

This is useful for wrapping helper objects in higher-level helpers or extending a helper to provide extra functionality without having to resort to subclassing.

The “proxying” is actually just copying the proxyable attribute onto the target.

>>> class AddHelper(object):
...     def __init__(self, number):
...         self._number = number
...
...     @proxyable
...     def add_number(self, number):
...         return self._number + number

>>> class OtherHelper(object):
...     def __init__(self, number):
...         self._adder = AddHelper(number)
...         generate_proxies(self, self._adder)
...
...     @proxyable
...     def say_hello(self):
...         return "hello"

>>> other_helper = OtherHelper(3)
>>> other_helper.say_hello()
'hello'
>>> other_helper.add_number(2)
5

vumi.tests.helpers.success_result_of(d)¶: We can’t necessarily use TestCase.successResultOf because our Twisted might not be new enough. This is a standalone copy with some minor message differences.

vumi.tests.helpers.get_timeout()¶

Look up the test timeout in the VUMI_TEST_TIMEOUT environment variable.

A default of 5 seconds is used if there isn’t one there.

vumi.tests.helpers.get_stack_trace(exclude_last=0)¶

Get a stack trace that can be stored and referred to later.

The inside of this function is excluded from the stack trace, because it’s not relevant. Additionally, all entries prior to the first occurrence of “twisted/trial/_asynctest.py” or “django/test/testcases.py” are removed to avoid unnecessary test runner noise.

Parameters:	exclude_last (int) – Number of entries to remove from the end of the stack trace. Use this to get rid of wrapper functions or implementation details irrelevant to the purpose of the stack trace.
Returns:	A list of strings, each representing a stack frame, in the same format as `traceback.format_stack()`.

class vumi.tests.helpers.MessageHelper(transport_name='sphex', transport_type='sms', mobile_addr='+41791234567', transport_addr='9292')¶

Bases: object

Test helper for constructing various messages.

This helper does no setup or cleanup. It takes the following parameters, which are used as defaults for message fields:

Parameters:	transport_name (str) – Default value for `transport_name` on all messages. transport_type (str) – Default value for `transport_type` on all messages. mobile_addr (str) – Default value for `from_addr` on inbound messages and `to_addr` on outbound messages. transport_addr (str) – Default value for `to_addr` on inbound messages and `from_addr` on outbound messages.

make_inbound(content, from_addr=DEFAULT, to_addr=DEFAULT, **kw)¶

Construct an inbound TransportUserMessage.

This is a convenience wrapper around make_user_message() and just sets to_addr and from_addr appropriately for an inbound message.

make_outbound(content, from_addr=DEFAULT, to_addr=DEFAULT, **kw)¶

Construct an outbound TransportUserMessage.

This is a convenience wrapper around make_user_message() and just sets to_addr and from_addr appropriately for an outbound message.

make_user_message(content, from_addr, to_addr, group=None, session_event=None, transport_type=DEFAULT, transport_name=DEFAULT, transport_metadata=DEFAULT, helper_metadata=DEFAULT, endpoint=DEFAULT, **kw)¶

Construct a TransportUserMessage.

This method is the underlying implementation for make_inbound() and make_outbound() and those should be used instead where they apply.

The only real difference between using this method and constructing a message object directly is that this method provides sensible defaults for most fields and sets the routing endpoint (if provided) in a more convenient way.

The following parameters are mandatory:

Parameters:	content (str) – Message `content` field. from_addr (str) – Message `from_addr` field. to_addr (str) – Message `to_addr` field.

The following parameters override default values for the message fields of the same name:

Parameters:	group (str) – Default `None`. session_event (str) – Default `None`. transport_type (str) – Default `transport_type`. transport_name (str) – Default `transport_name`. transport_metadata (dict) – Default `{}`. helper_metadata (dict) – Default `{}`.

The following parameter is special:

Parameters:	endpoint (str) – If specified, the routing endpoint on the message is set by calling `TransportUserMessage.set_routing_endpoint()`.

All other keyword args are passed to the TransportUserMessage constructor.

make_event(event_type, user_message_id, transport_type=DEFAULT, transport_name=DEFAULT, transport_metadata=DEFAULT, endpoint=DEFAULT, **kw)¶

Construct a TransportEvent.

This method is the underlying implementation for make_ack(), make_nack() and make_delivery_report(). Those should be used instead where they apply.

The only real difference between using this method and constructing an event object directly is that this method provides sensible defaults for most fields and sets the routing endpoint (if provided) in a more convenient way.

The following parameters are mandatory:

Parameters:	event_type (str) – Event `event_type` field. user_message_id (str) – Event `user_message_id` field.

Any fields required by a particular event type (such as sent_message_id for ack events) are also mandatory.

The following parameters override default values for the event fields of the same name:

Parameters:	transport_type (str) – Default `transport_type`. transport_name (str) – Default `transport_name`. transport_metadata (dict) – Default `{}`.

The following parameter is special:

Parameters:	endpoint (str) – If specified, the routing endpoint on the event is set by calling `TransportUserMessage.set_routing_endpoint()`.

All other keyword args are passed to the TransportEvent constructor.

make_ack(msg=None, sent_message_id=DEFAULT, **kw)¶

Construct an ‘ack’ TransportEvent.

Parameters:	msg – `TransportUserMessage` instance the event is for. If `None`, this method will call `make_outbound()` to get one. sent_message_id (str) – If this isn’t provided, `msg['message_id']` will be used.

All remaining keyword params are passed to make_event().

make_nack(msg=None, nack_reason=DEFAULT, **kw)¶

Construct a ‘nack’ TransportEvent.

Parameters:	msg – `TransportUserMessage` instance the event is for. If `None`, this method will call `make_outbound()` to get one. nack_reason (str) – If this isn’t provided, a suitable excuse will be used.

All remaining keyword params are passed to make_event().

make_delivery_report(msg=None, delivery_status=DEFAULT, **kw)¶

Construct a ‘delivery_report’ TransportEvent.

Parameters:	msg – `TransportUserMessage` instance the event is for. If `None`, this method will call `make_outbound()` to get one. delivery_status (str) – If this isn’t provided, `"delivered"` will be used.

All remaining keyword params are passed to make_event().

make_reply(msg, content, **kw)¶

Construct a reply TransportUserMessage.

This literally just calls msg.reply(content, **kw). It is included for completeness and symmetry with MessageDispatchHelper.make_dispatch_reply().

make_status(**kw)¶: Construct a TransportStatus.

class vumi.tests.helpers.WorkerHelper(connector_name=None, broker=None, status_connector_name=None)¶

Bases: object

Test helper for creating workers and dispatching messages.

This helper does no setup, but it waits for pending message deliveries and the stops all workers it knows about during cleanup. It takes the following parameters:

Parameters:	connector_name (str) – Default value for `connector_name` on all message broker operations. If `None`, the connector name must be provided for each operation. broker – The message broker to use internally. This should be an instance of `FakeAMQPBroker` if it is provided, but most of the time the default of `None` should be used to have the helper create its own broker.

cleanup(*args, **kwargs)¶: Wait for any pending message deliveries and stop all workers.

cleanup_worker(worker)¶: Clean up a particular worker manually and remove it from the helper’s cleanup list. This should only be called with workers that are already in the helper’s cleanup list.

classmethod get_fake_amqp_client(broker)¶

Wrap a fake broker in an fake client.

The broker parameter is mandatory because it’s important that cleanup happen. If None is passed in explicitly, a new broker object will be created.

classmethod get_worker_raw(worker_class, config, broker=None)¶

Create and return an instance of a vumi worker.

This doesn’t start the worker and it doesn’t add it to any cleanup machinery. In most cases, you want get_worker() instead.

get_worker(worker_class, config, start=True)¶

Create and return an instance of a vumi worker.

Parameters:	worker_class – The worker class to instantiate. config – Config dict. start – `True` to start the worker (default), `False` otherwise.

get_dispatched(connector_name, name, message_class)¶

Get messages dispatched to a routing key.

The more specific get_dispatched_events(), get_dispatched_inbound(), and get_dispatched_outbound() wrapper methods should be used instead where they apply.

Parameters:	connector_name (str) – The connector name, which is used as the routing key prefix. name (str) – The routing key suffix, generally `"event"`, `"inbound"`, or `"outbound"`. message_class – The message class to wrap the raw message data in. This should probably be `TransportUserMessage` or `TransportEvent`.

clear_all_dispatched()¶: Clear all dispatched messages from the broker.

get_dispatched_events(connector_name=None)¶

Get events dispatched to a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A list of `TransportEvent` instances.

get_dispatched_inbound(connector_name=None)¶

Get inbound messages dispatched to a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A list of `TransportUserMessage` instances.

get_dispatched_outbound(connector_name=None)¶

Get outbound messages dispatched to a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A list of `TransportUserMessage` instances.

get_dispatched_statuses(connector_name=None)¶

Get statuses dispatched to a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default status connector name for the helper instance will be used.
Returns:	A list of `TransportStatus` instances.

wait_for_dispatched_events(amount=None, connector_name=None)¶

Wait for events dispatched to a connector.

Parameters:	amount (int) – Number of messages to wait for. If `None`, this will wait for the end of the current delivery run instead of a specific number of messages. connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A `Deferred` that fires with a list of `TransportEvent` instances.

wait_for_dispatched_inbound(amount=None, connector_name=None)¶

Wait for inbound messages dispatched to a connector.

Parameters:	amount (int) – Number of messages to wait for. If `None`, this will wait for the end of the current delivery run instead of a specific number of messages. connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A `Deferred` that fires with a list of `TransportUserMessage` instances.

wait_for_dispatched_outbound(amount=None, connector_name=None)¶

Wait for outbound messages dispatched to a connector.

Parameters:	amount (int) – Number of messages to wait for. If `None`, this will wait for the end of the current delivery run instead of a specific number of messages. connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A `Deferred` that fires with a list of `TransportUserMessage` instances.

wait_for_dispatched_statuses(amount=None, connector_name=None)¶

Wait for statuses dispatched to a connector.

Parameters:	amount (int) – Number of messages to wait for. If `None`, this will wait for the end of the current delivery run instead of a specific number of messages. connector_name (str) – Connector name. If `None`, the default status connector name for the helper instance will be used.
Returns:	A `Deferred` that fires with a list of `TransportEvent` instances.

clear_dispatched_events(connector_name=None)¶

Clear dispatched events for a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.

clear_dispatched_inbound(connector_name=None)¶

Clear dispatched inbound messages for a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.

clear_dispatched_outbound(connector_name=None)¶

Clear dispatched outbound messages for a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.

clear_dispatched_statuses(connector_name=None)¶

Clear dispatched statuses for a connector.

Parameters:	connector_name (str) – Connector name. If `None`, the default status connector name for the helper instance will be used.

dispatch_raw(routing_key, message, exchange='vumi')¶

Dispatch a message to the specified routing key.

The more specific dispatch_inbound(), dispatch_outbound(), and dispatch_event() wrapper methods should be used instead where they apply.

Parameters:	routing_key (str) – Routing key to dispatch the message to. message – Message to dispatch. exchange (str) – AMQP exchange to dispatch the message to. Defaults to `"vumi"`
Returns:	A `Deferred` that fires when all messages have been delivered.

dispatch_inbound(message, connector_name=None)¶

Dispatch an inbound message.

Parameters:	message – Message to dispatch. Should be a `TransportUserMessage` instance. connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A `Deferred` that fires when all messages have been delivered.

dispatch_outbound(message, connector_name=None)¶

Dispatch an outbound message.

Parameters:	message – Message to dispatch. Should be a `TransportUserMessage` instance. connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A `Deferred` that fires when all messages have been delivered.

dispatch_event(message, connector_name=None)¶

Dispatch an event.

Parameters:	message – Message to dispatch. Should be a `TransportEvent` instance. connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A `Deferred` that fires when all messages have been delivered.

dispatch_status(message, connector_name=None)¶

Dispatch a status.

Parameters:	message – Message to dispatch. Should be a `TransportStatus` instance. connector_name (str) – Connector name. If `None`, the default status connector name for the helper instance will be used.
Returns:	A `Deferred` that fires when all messages have been delivered.

kick_delivery()¶

Trigger delivery of messages by the broker.

This is generally called internally by anything that sends a message.

Returns:	A `Deferred` that fires when all messages have been delivered.

get_dispatched_metrics()¶

Get dispatched metrics.

The list of datapoints from each dispatched metrics message is returned.

wait_for_dispatched_metrics()¶

Get dispatched metrics after waiting for any pending deliveries.

The list of datapoints from each dispatched metrics message is returned.

clear_dispatched_metrics()¶: Clear dispatched metrics messages from the broker.

class vumi.tests.helpers.MessageDispatchHelper(msg_helper, worker_helper)¶

Bases: object

Helper for creating and immediately dispatching messages.

This builds on top of MessageHelper and WorkerHelper.

It does not allow dispatching to nonstandard connectors. If you need to do that, either use MessageHelper and WorkerHelper directly or build a second MessageDispatchHelper with a second WorkerHelper.

Parameters:	msg_helper – A `MessageHelper` instance. worker_helper – A `WorkerHelper` instance.

make_dispatch_inbound(*args, **kw)¶

Construct and dispatch an inbound message.

This is a wrapper around MessageHelper.make_inbound() (to which all parameters are passed) and WorkerHelper.dispatch_inbound().

Returns:	A `Deferred` that fires with the constructed message once it has been dispatched.

make_dispatch_outbound(*args, **kw)¶

Construct and dispatch an outbound message.

This is a wrapper around MessageHelper.make_outbound() (to which all parameters are passed) and WorkerHelper.dispatch_outbound().

Returns:	A `Deferred` that fires with the constructed message once it has been dispatched.

make_dispatch_ack(*args, **kw)¶

Construct and dispatch an ack event.

This is a wrapper around MessageHelper.make_ack() (to which all parameters are passed) and WorkerHelper.dispatch_event().

Returns:	A `Deferred` that fires with the constructed event once it has been dispatched.

make_dispatch_nack(*args, **kw)¶

Construct and dispatch a nack event.

This is a wrapper around MessageHelper.make_nack() (to which all parameters are passed) and WorkerHelper.dispatch_event().

Returns:	A `Deferred` that fires with the constructed event once it has been dispatched.

make_dispatch_delivery_report(*args, **kw)¶

Construct and dispatch a delivery report event.

This is a wrapper around MessageHelper.make_delivery_report() (to which all parameters are passed) and WorkerHelper.dispatch_event().

Returns:	A `Deferred` that fires with the constructed event once it has been dispatched.

make_dispatch_reply(*args, **kw)¶

Construct and dispatch a reply message.

This is a wrapper around MessageHelper.make_reply() (to which all parameters are passed) and WorkerHelper.dispatch_outbound().

Returns:	A `Deferred` that fires with the constructed message once it has been dispatched.

make_dispatch_status(*args, **kw)¶

Construct and dispatch a status.

This is a wrapper around MessageHelper.make_status() (to which all parameters are passed) and WorkerHelper.dispatch_status().

Returns:	A `Deferred` that fires with the constructed message once it has been dispatched.

class vumi.tests.helpers.RiakDisabledForTest¶

Bases: object

Placeholder object for a disabled riak config.

This class exists to throw a meaningful error when trying to use Riak in a test that disallows it. We can’t do this from inside the Riak setup infrastructure, because that would be very invasive for something that only really matters for tests.

vumi.tests.helpers.import_skip(exc, *expected)¶

Raise SkipTest if the provided ImportError matches a module name in expected, otherwise reraise the ImportError.

This is useful for skipping tests that require optional dependencies which might not be present.

vumi.tests.helpers.skiptest(reason)¶

Decorate a test that should be skipped with a reason.

NOTE: Don’t import this as skip, because that will cause trial to skip: the entire module that imports it.

vumi.tests.helpers.maybe_async(sync_attr)¶

Decorate a method that may be sync or async.

This redecorates with the either @inlineCallbacks or @flatten_generator, depending on the value of sync_attr.

vumi.tests.helpers.maybe_async_return(value, maybe_deferred)¶

Return value or a deferred that fires with it.

This is useful in cases where we’re performing a potentially async operation but don’t necessarily have enough information to use maybe_async.

exception vumi.tests.helpers.PersistenceHelperError¶

Bases: exceptions.Exception

Exception thrown by a PersistenceHelper when it sees something wrong.

class vumi.tests.helpers.PersistenceHelper(use_riak=False, is_sync=False, assert_closed=False)¶

Bases: object

Test helper for managing persistent storage.

This helper manages Riak and Redis clients and configs and cleans up after them. It does no setup, but its cleanup may take a while if there’s a lot in Riak.

All configs for objects that build Riak or Redis clients must be passed through mk_config().

Parameters:	use_riak (bool) – Pass `True` if Riak is desired, otherwise it will be disabled in the generated config parameters. is_sync (bool) – Pass `True` if synchronous Riak and Redis clients are desired, otherwise asynchronous ones will be built. This only applies to clients built by this helper, not those built by other objects using configs from this helper.

get_riak_manager(config=None)¶

Build and return a Riak manager.

Parameters:	config (dict) – Riak manager config. (Not a complete worker config.) If `None`, the one used by `mk_config()` will be used.
Returns:	A `RiakManager` or `TxRiakManager`, depending on the value of `is_sync`.

record_load_and_store(riak_manager, loads, stores)¶

Patch a Riak manager to capture load and store operations.

Parameters:	riak_manager – The manager object to patch. loads (list) – A list to append the keys of loaded objects to. stores (list) – A list to append the keys of stored objects to.

get_redis_manager(config=None)¶

Build and return a Redis manager.

This will be backed by an in-memory fake unless the VUMITEST_REDIS_DB environment variable is set.

Parameters:	config (dict) – Redis manager config. (Not a complete worker config.) If `None`, the one used by `mk_config()` will be used.
Returns:	A `RedisManager` or `TxRedisManager`, depending on the value of `is_sync`.

mk_config(config)¶

Return a copy of config with the riak_manager and redis_manager fields overridden.

All configs for things that create Riak or Redis clients should be passed through this method.

Application worker test helpers¶

class vumi.application.tests.helpers.ApplicationHelper(application_class, use_riak=False, **msg_helper_args)¶

Bases: object

Test helper for application workers.

This helper construct and wraps several lower-level helpers and provides higher-level functionality for app worker tests.

Parameters:	application_class – The worker class for the application being tested. use_riak (bool) – Set to `True` if the test requires Riak. This is passed to the underlying `PersistenceHelper`. **msg_helper_args – All other keyword params are passed to the underlying `MessageHelper`.

get_application(config, cls=None, start=True)¶

Get an instance of a worker class.

Parameters:	config – Config dict. cls – The Application class to instantiate. Defaults to `application_class` start – True to start the application (default), False otherwise.

Some default config values are helpfully provided in the interests of reducing boilerplate:

transport_name defaults to self.transport_name

vumi.application.tests.helpers.find_nodejs_or_skip_test(worker_class)¶: Find the node.js executable by checking the VUMI_TEST_NODE_PATH envvar and falling back to the provided worker’s own detection method. If no executable is found, SkipTest is raised.

Transport worker test helpers¶

class vumi.transports.tests.helpers.TransportHelper(transport_class, use_riak=False, **msg_helper_args)¶

Bases: object

Test helper for transport workers.

This helper construct and wraps several lower-level helpers and provides higher-level functionality for transport tests.

Parameters:	transport_class – The worker class for the transport being tested. use_riak (bool) – Set to `True` if the test requires Riak. This is passed to the underlying `PersistenceHelper`. **msg_helper_args – All other keyword params are passed to the underlying `MessageHelper`.

get_transport(config, cls=None, start=True)¶

Get an instance of a transport class.

Parameters:	config – Config dict. cls – The transport class to instantiate. Defaults to `transport_class` start – True to start the transport (default), False otherwise.

Some default config values are helpfully provided in the interests of reducing boilerplate:

transport_name defaults to self.transport_name

get_dispatched_failures(connector_name=None)¶

Get failures dispatched by a transport.

Parameters:	connector_name (str) – Connector name. If `None`, the default connector name for the helper instance will be used.
Returns:	A list of `FailureMessage` instances.

exception vumi.transports.httprpc.tests.helpers.HttpRpcTransportHelperError¶

Bases: vumi.errors.VumiError

Error raised when the HttpRpcTransportHelper encouters an error.

class vumi.transports.httprpc.tests.helpers.HttpRpcTransportHelper(transport_class, use_riak=False, request_defaults=None, **msg_helper_args)¶

Bases: object

Test helper for subclasses of HttpRpcTransport.

Adds support for making HTTP requests to the HTTP RPC transport to the base TransportHelper.

Parameters:	request_defaults (dict) – Default URL parameters for HTTP requests.

Other parameters are the same as for TransportHelper.

mk_request_raw(suffix='', params=None, data=None, method='GET')¶

Make an HTTP request, ignoring this helper’s request_defaults.

Parameters:	suffix (str) – Suffix to add to the transport’s URL. params (dict) – A dictionary of URL parameters to append to the URL as a query string or None for no URL parameters. data (str) – Request body or None for no request body. method (str) – HTTP method to use for the request.
Raises:	HttpRpcTransportHelperError – When invoked before calling `get_transport()`.

mk_request(_suffix='', _data=None, _method='GET', **kw)¶

Make an HTTP request.

Parameters:	_suffix (str) – Suffix to add to the transport’s URL. _data (str) – Request body or None for no request body. _method (str) – HTTP method to use for the request. **kw – URL query string parameters.
Raises:	HttpRpcTransportHelperError – When invoked before calling `get_transport()`.

The _ prefixes on the function parameter names are to make accidental clashes with URL query parameter names less likely.

Dispatcher worker test helpers¶

class vumi.dispatchers.tests.helpers.DispatcherHelper(dispatcher_class, use_riak=False, **msg_helper_args)¶

Bases: object

Test helper for dispatcher workers.

This helper construct and wraps several lower-level helpers and provides higher-level functionality for dispatcher tests.

Parameters:	dispatcher_class – The worker class for the dispatcher being tested. use_riak (bool) – Set to `True` if the test requires Riak. This is passed to the underlying `PersistenceHelper`. **msg_helper_args – All other keyword params are passed to the underlying `MessageHelper`.

get_dispatcher(config, cls=None, start=True)¶

Get an instance of a dispatcher class.

Parameters:	config (dict) – Config dict. cls – The transport class to instantiate. Defaults to `dispatcher_class` start (bool) – `True` to start the dispatcher (default), `False` otherwise.

get_connector_helper(connector_name)¶: Construct a DispatcherConnectorHelper for the provided connector_name.

class vumi.dispatchers.tests.helpers.DispatcherConnectorHelper(dispatcher_helper, connector_name)¶

Bases: object

Subset of WorkerHelper and MessageDispatchHelper functionality for a specific connector. This should only be created with DispatcherHelper.get_connector_helper().