| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172 |
- from __future__ import absolute_import, unicode_literals
- from itertools import count
- from case import ContextMock, Mock
- from kombu.transport import base
- from kombu.utils import json
- def PromiseMock(*args, **kwargs):
- m = Mock(*args, **kwargs)
- def on_throw(exc=None, *args, **kwargs):
- if exc:
- raise exc
- raise
- m.throw.side_effect = on_throw
- m.set_error_state.side_effect = on_throw
- m.throw1.side_effect = on_throw
- return m
- class MockPool(object):
- def __init__(self, value=None):
- self.value = value or ContextMock()
- def acquire(self, **kwargs):
- return self.value
- class Message(base.Message):
- def __init__(self, *args, **kwargs):
- self.throw_decode_error = kwargs.get('throw_decode_error', False)
- super(Message, self).__init__(*args, **kwargs)
- def decode(self):
- if self.throw_decode_error:
- raise ValueError("can't decode message")
- return super(Message, self).decode()
- class Channel(base.StdChannel):
- open = True
- throw_decode_error = False
- _ids = count(1)
- def __init__(self, connection):
- self.connection = connection
- self.called = []
- self.deliveries = count(1)
- self.to_deliver = []
- self.events = {'basic_return': set()}
- self.channel_id = next(self._ids)
- def _called(self, name):
- self.called.append(name)
- def __contains__(self, key):
- return key in self.called
- def exchange_declare(self, *args, **kwargs):
- self._called('exchange_declare')
- def prepare_message(self, body, priority=0, content_type=None,
- content_encoding=None, headers=None, properties={}):
- self._called('prepare_message')
- return {'body': body,
- 'headers': headers,
- 'properties': properties,
- 'priority': priority,
- 'content_type': content_type,
- 'content_encoding': content_encoding}
- def basic_publish(self, message, exchange='', routing_key='',
- mandatory=False, immediate=False, **kwargs):
- self._called('basic_publish')
- return message, exchange, routing_key
- def exchange_delete(self, *args, **kwargs):
- self._called('exchange_delete')
- def queue_declare(self, *args, **kwargs):
- self._called('queue_declare')
- def queue_bind(self, *args, **kwargs):
- self._called('queue_bind')
- def queue_unbind(self, *args, **kwargs):
- self._called('queue_unbind')
- def queue_delete(self, queue, if_unused=False, if_empty=False, **kwargs):
- self._called('queue_delete')
- def basic_get(self, *args, **kwargs):
- self._called('basic_get')
- try:
- return self.to_deliver.pop()
- except IndexError:
- pass
- def queue_purge(self, *args, **kwargs):
- self._called('queue_purge')
- def basic_consume(self, *args, **kwargs):
- self._called('basic_consume')
- def basic_cancel(self, *args, **kwargs):
- self._called('basic_cancel')
- def basic_ack(self, *args, **kwargs):
- self._called('basic_ack')
- def basic_recover(self, requeue=False):
- self._called('basic_recover')
- def exchange_bind(self, *args, **kwargs):
- self._called('exchange_bind')
- def exchange_unbind(self, *args, **kwargs):
- self._called('exchange_unbind')
- def close(self):
- self._called('close')
- def message_to_python(self, message, *args, **kwargs):
- self._called('message_to_python')
- return Message(body=json.dumps(message),
- channel=self,
- delivery_tag=next(self.deliveries),
- throw_decode_error=self.throw_decode_error,
- content_type='application/json',
- content_encoding='utf-8')
- def flow(self, active):
- self._called('flow')
- def basic_reject(self, delivery_tag, requeue=False):
- if requeue:
- return self._called('basic_reject:requeue')
- return self._called('basic_reject')
- def basic_qos(self, prefetch_size=0, prefetch_count=0,
- apply_global=False):
- self._called('basic_qos')
- class Connection(object):
- connected = True
- def __init__(self, client):
- self.client = client
- def channel(self):
- return Channel(self)
- class Transport(base.Transport):
- def establish_connection(self):
- return Connection(self.client)
- def create_channel(self, connection):
- return connection.channel()
- def drain_events(self, connection, **kwargs):
- return 'event'
- def close_connection(self, connection):
- connection.connected = False
|