Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_upstream_blackhole(self, container, publish, toxiproxy):
""" Verify we detect and recover from sockets losing data.
This failure mode means that all data sent from the consumer to the
rabbit broker is lost, but the socket remains open.
Heartbeats sent from the consumer are not received by the broker. After
two beats are missed the broker closes the connection, and subsequent
reads from the socket raise a socket.error, so the connection is
re-established.
"""
queue_consumer = get_extension(container, QueueConsumer)
def reset(args, kwargs, result, exc_info):
toxiproxy.reset_timeout()
return True
with patch_wait(queue_consumer, 'on_connection_error', callback=reset):
toxiproxy.set_timeout(timeout=0)
# connection re-established
msg = "foo"
with entrypoint_waiter(container, 'echo') as result:
publish(msg)
assert result.get() == msg
def test_dependency_call_lifecycle_errors(
container_factory, rabbit_config, method_name):
container = container_factory(ExampleService, rabbit_config)
container.start()
dependency = get_extension(container, EventDispatcher)
with patch.object(dependency, method_name, autospec=True) as method:
err = "error in {}".format(method_name)
method.side_effect = Exception(err)
# use a standalone rpc proxy to call exampleservice.task()
with ServiceRpcProxy("exampleservice", rabbit_config) as proxy:
# proxy.task() will hang forever because it generates an error
proxy.task.call_async()
# verify that the error bubbles up to container.wait()
with pytest.raises(Exception) as exc_info:
container.wait()
assert str(exc_info.value) == err
res = result.get()
# with custom timeout
with entrypoint_waiter(container, 'handle_event', timeout=5):
... # action that dispatches event
# with callback that waits until entrypoint stops raising
def callback(worker_ctx, result, exc_info):
if exc_info is None:
return True
with entrypoint_waiter(container, 'handle_event', callback=callback):
... # action that dispatches event
"""
if not get_extension(container, Entrypoint, method_name=method_name):
raise RuntimeError("{} has no entrypoint `{}`".format(
container.service_name, method_name))
class Result(WaitResult):
worker_ctx = None
def send(self, worker_ctx, result, exc_info):
self.worker_ctx = worker_ctx
super(Result, self).send(result, exc_info)
waiter_callback = callback
waiter_result = Result()
def on_worker_result(worker_ctx, result, exc_info):
complete = False
if worker_ctx.entrypoint.method_name == method_name:
def test_dependency_uniqueness(container_factory):
c1 = container_factory(Service, config={})
c2 = container_factory(Service, config={})
# dependencyprovider declarations are identical between containers
assert c1.service_cls.dep == c2.service_cls.dep
# dependencyprovider instances are different between containers
dep1 = get_extension(c1, SimpleDependencyProvider)
dep2 = get_extension(c2, SimpleDependencyProvider)
assert dep1 != dep2
class Service:
name = 'sample'
@rtm.handle_event(expected_exceptions=Boom)
def handle_event(self, event):
pass
@rtm.handle_message(expected_exceptions=Boom)
def handle_message(self, event, message):
pass
container = container_factory(Service, config)
event_handler = get_extension(container, rtm.RTMEventHandlerEntrypoint)
assert event_handler.expected_exceptions == Boom
message_handler = get_extension(
container, rtm.RTMMessageHandlerEntrypoint)
assert message_handler.expected_exceptions == Boom
def factory(config=None, bot_name=None):
class Service(object):
name = "service"
slack_api = Slack(bot_name)
@dummy
def dummy(self):
pass
container = ServiceContainer(Service, config or default_config)
containers.append(container)
return get_extension(container, Slack)
def test_worker_life_cycle(container):
spam_dep = get_extension(container, DependencyProvider)
ham_dep = get_extension(container, Entrypoint, method_name="ham")
egg_dep = get_extension(container, Entrypoint, method_name="egg")
handle_result = Mock()
handle_result.side_effect = (
lambda worker_ctx, res, exc_info: (res, exc_info))
ham_worker_ctx = container.spawn_worker(
ham_dep, [], {}, handle_result=handle_result)
container._worker_pool.waitall()
egg_worker_ctx = container.spawn_worker(
egg_dep, [], {}, handle_result=handle_result)
container._worker_pool.waitall()
assert spam_dep.calls == [
def test_default(self, service_cls, container_factory, rabbit_config):
container = container_factory(service_cls, rabbit_config)
container.start()
queue_consumer = get_extension(container, QueueConsumer)
assert queue_consumer.connection.heartbeat == DEFAULT_HEARTBEAT
The name of the entrypoint decorated method on the service class
context_data : dict
Context data to provide for the call, e.g. a language, auth
token or session.
timeout : int
Maximum seconds to wait
**Usage**
To verify that `ServiceX` and `ServiceY` are compatible, make an
integration test that checks their interaction:
.. literalinclude:: ../examples/testing/integration_x_y_test.py
"""
entrypoint = get_extension(container, Entrypoint, method_name=method_name)
if entrypoint is None:
raise ExtensionNotFound(
"No entrypoint for '{}' found on container {}.".format(
method_name, container))
def hook(*args, **kwargs):
hook_result = event.Event()
def wait_for_entrypoint():
try:
with entrypoint_waiter(
container, method_name,
timeout=timeout
) as waiter_result:
container.spawn_worker(
entrypoint, args, kwargs,
def test_config_value(
self, heartbeat, service_cls, container_factory, rabbit_config
):
rabbit_config[HEARTBEAT_CONFIG_KEY] = heartbeat
container = container_factory(service_cls, rabbit_config)
container.start()
queue_consumer = get_extension(container, QueueConsumer)
assert queue_consumer.connection.heartbeat == heartbeat