123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540 |
- from __future__ import absolute_import
-
- import warnings
-
- from kombu import Connection
- from kombu.exceptions import ResourceError, ChannelError
- from kombu.transport import virtual
- from kombu.utils import uuid
- from kombu.compression import compress
-
- from kombu.tests.case import Case, Mock, patch, redirect_stdouts
-
-
- def client(**kwargs):
- return Connection(transport='kombu.transport.virtual:Transport', **kwargs)
-
-
- def memory_client():
- return Connection(transport='memory')
-
-
- class test_BrokerState(Case):
-
- def test_constructor(self):
- s = virtual.BrokerState()
- self.assertTrue(hasattr(s, 'exchanges'))
- self.assertTrue(hasattr(s, 'bindings'))
-
- t = virtual.BrokerState(exchanges=16, bindings=32)
- self.assertEqual(t.exchanges, 16)
- self.assertEqual(t.bindings, 32)
-
-
- class test_QoS(Case):
-
- def setUp(self):
- self.q = virtual.QoS(client().channel(), prefetch_count=10)
-
- def tearDown(self):
- self.q._on_collect.cancel()
-
- def test_constructor(self):
- self.assertTrue(self.q.channel)
- self.assertTrue(self.q.prefetch_count)
- self.assertFalse(self.q._delivered.restored)
- self.assertTrue(self.q._on_collect)
-
- @redirect_stdouts
- def test_can_consume(self, stdout, stderr):
- _restored = []
-
- class RestoreChannel(virtual.Channel):
- do_restore = True
-
- def _restore(self, message):
- _restored.append(message)
-
- self.assertTrue(self.q.can_consume())
- for i in range(self.q.prefetch_count - 1):
- self.q.append(i, uuid())
- self.assertTrue(self.q.can_consume())
- self.q.append(i + 1, uuid())
- self.assertFalse(self.q.can_consume())
-
- tag1 = next(iter(self.q._delivered))
- self.q.ack(tag1)
- self.assertTrue(self.q.can_consume())
-
- tag2 = uuid()
- self.q.append(i + 2, tag2)
- self.assertFalse(self.q.can_consume())
- self.q.reject(tag2)
- self.assertTrue(self.q.can_consume())
-
- self.q.channel = RestoreChannel(self.q.channel.connection)
- tag3 = uuid()
- self.q.append(i + 3, tag3)
- self.q.reject(tag3, requeue=True)
- self.q._flush()
- self.q.restore_unacked_once()
- self.assertListEqual(_restored, [11, 9, 8, 7, 6, 5, 4, 3, 2, 1])
- self.assertTrue(self.q._delivered.restored)
- self.assertFalse(self.q._delivered)
-
- self.q.restore_unacked_once()
- self.q._delivered.restored = False
- self.q.restore_unacked_once()
-
- self.assertTrue(stderr.getvalue())
- self.assertFalse(stdout.getvalue())
-
- self.q.restore_at_shutdown = False
- self.q.restore_unacked_once()
-
- def test_get(self):
- self.q._delivered['foo'] = 1
- self.assertEqual(self.q.get('foo'), 1)
-
-
- class test_Message(Case):
-
- def test_create(self):
- c = client().channel()
- data = c.prepare_message('the quick brown fox...')
- tag = data['properties']['delivery_tag'] = uuid()
- message = c.message_to_python(data)
- self.assertIsInstance(message, virtual.Message)
- self.assertIs(message, c.message_to_python(message))
- if message.errors:
- message._reraise_error()
-
- self.assertEqual(message.body,
- 'the quick brown fox...'.encode('utf-8'))
- self.assertTrue(message.delivery_tag, tag)
-
- def test_create_no_body(self):
- virtual.Message(Mock(), {
- 'body': None,
- 'properties': {'delivery_tag': 1}})
-
- def test_serializable(self):
- c = client().channel()
- body, content_type = compress('the quick brown fox...', 'gzip')
- data = c.prepare_message(body, headers={'compression': content_type})
- tag = data['properties']['delivery_tag'] = uuid()
- message = c.message_to_python(data)
- dict_ = message.serializable()
- self.assertEqual(dict_['body'],
- 'the quick brown fox...'.encode('utf-8'))
- self.assertEqual(dict_['properties']['delivery_tag'], tag)
- self.assertFalse('compression' in dict_['headers'])
-
-
- class test_AbstractChannel(Case):
-
- def test_get(self):
- with self.assertRaises(NotImplementedError):
- virtual.AbstractChannel()._get('queue')
-
- def test_put(self):
- with self.assertRaises(NotImplementedError):
- virtual.AbstractChannel()._put('queue', 'm')
-
- def test_size(self):
- self.assertEqual(virtual.AbstractChannel()._size('queue'), 0)
-
- def test_purge(self):
- with self.assertRaises(NotImplementedError):
- virtual.AbstractChannel()._purge('queue')
-
- def test_delete(self):
- with self.assertRaises(NotImplementedError):
- virtual.AbstractChannel()._delete('queue')
-
- def test_new_queue(self):
- self.assertIsNone(virtual.AbstractChannel()._new_queue('queue'))
-
- def test_has_queue(self):
- self.assertTrue(virtual.AbstractChannel()._has_queue('queue'))
-
- def test_poll(self):
-
- class Cycle(object):
- called = False
-
- def get(self):
- self.called = True
- return True
-
- cycle = Cycle()
- self.assertTrue(virtual.AbstractChannel()._poll(cycle))
- self.assertTrue(cycle.called)
-
-
- class test_Channel(Case):
-
- def setUp(self):
- self.channel = client().channel()
-
- def tearDown(self):
- if self.channel._qos is not None:
- self.channel._qos._on_collect.cancel()
-
- def test_exceeds_channel_max(self):
- c = client()
- t = c.transport
- avail = t._avail_channel_ids = Mock(name='_avail_channel_ids')
- avail.pop.side_effect = IndexError()
- with self.assertRaises(ResourceError):
- virtual.Channel(t)
-
- def test_exchange_bind_interface(self):
- with self.assertRaises(NotImplementedError):
- self.channel.exchange_bind('dest', 'src', 'key')
-
- def test_exchange_unbind_interface(self):
- with self.assertRaises(NotImplementedError):
- self.channel.exchange_unbind('dest', 'src', 'key')
-
- def test_queue_unbind_interface(self):
- with self.assertRaises(NotImplementedError):
- self.channel.queue_unbind('dest', 'ex', 'key')
-
- def test_management(self):
- m = self.channel.connection.client.get_manager()
- self.assertTrue(m)
- m.get_bindings()
- m.close()
-
- def test_exchange_declare(self):
- c = self.channel
-
- with self.assertRaises(ChannelError):
- c.exchange_declare('test_exchange_declare', 'direct',
- durable=True, auto_delete=True, passive=True)
- c.exchange_declare('test_exchange_declare', 'direct',
- durable=True, auto_delete=True)
- c.exchange_declare('test_exchange_declare', 'direct',
- durable=True, auto_delete=True, passive=True)
- self.assertIn('test_exchange_declare', c.state.exchanges)
- # can declare again with same values
- c.exchange_declare('test_exchange_declare', 'direct',
- durable=True, auto_delete=True)
- self.assertIn('test_exchange_declare', c.state.exchanges)
-
- # using different values raises NotEquivalentError
- with self.assertRaises(virtual.NotEquivalentError):
- c.exchange_declare('test_exchange_declare', 'direct',
- durable=False, auto_delete=True)
-
- def test_exchange_delete(self, ex='test_exchange_delete'):
-
- class PurgeChannel(virtual.Channel):
- purged = []
-
- def _purge(self, queue):
- self.purged.append(queue)
-
- c = PurgeChannel(self.channel.connection)
-
- c.exchange_declare(ex, 'direct', durable=True, auto_delete=True)
- self.assertIn(ex, c.state.exchanges)
- self.assertNotIn(ex, c.state.bindings) # no bindings yet
- c.exchange_delete(ex)
- self.assertNotIn(ex, c.state.exchanges)
-
- c.exchange_declare(ex, 'direct', durable=True, auto_delete=True)
- c.queue_declare(ex)
- c.queue_bind(ex, ex, ex)
- self.assertTrue(c.state.bindings[ex])
- c.exchange_delete(ex)
- self.assertNotIn(ex, c.state.bindings)
- self.assertIn(ex, c.purged)
-
- def test_queue_delete__if_empty(self, n='test_queue_delete__if_empty'):
- class PurgeChannel(virtual.Channel):
- purged = []
- size = 30
-
- def _purge(self, queue):
- self.purged.append(queue)
-
- def _size(self, queue):
- return self.size
-
- c = PurgeChannel(self.channel.connection)
- c.exchange_declare(n)
- c.queue_declare(n)
- c.queue_bind(n, n, n)
- # tests code path that returns if queue already bound.
- c.queue_bind(n, n, n)
-
- c.queue_delete(n, if_empty=True)
- self.assertIn(n, c.state.bindings)
-
- c.size = 0
- c.queue_delete(n, if_empty=True)
- self.assertNotIn(n, c.state.bindings)
- self.assertIn(n, c.purged)
-
- def test_queue_purge(self, n='test_queue_purge'):
-
- class PurgeChannel(virtual.Channel):
- purged = []
-
- def _purge(self, queue):
- self.purged.append(queue)
-
- c = PurgeChannel(self.channel.connection)
- c.exchange_declare(n)
- c.queue_declare(n)
- c.queue_bind(n, n, n)
- c.queue_purge(n)
- self.assertIn(n, c.purged)
-
- def test_basic_publish_unique_delivery_tags(self, n='test_uniq_tag'):
- c1 = memory_client().channel()
- c2 = memory_client().channel()
-
- for c in (c1, c2):
- c.exchange_declare(n)
- c.queue_declare(n)
- c.queue_bind(n, n, n)
- m1 = c1.prepare_message('George Costanza')
- m2 = c2.prepare_message('Elaine Marie Benes')
- c1.basic_publish(m1, n, n)
- c2.basic_publish(m2, n, n)
-
- r1 = c1.message_to_python(c1.basic_get(n))
- r2 = c2.message_to_python(c2.basic_get(n))
-
- self.assertNotEqual(r1.delivery_tag, r2.delivery_tag)
- with self.assertRaises(ValueError):
- int(r1.delivery_tag)
- with self.assertRaises(ValueError):
- int(r2.delivery_tag)
-
- def test_basic_publish__get__consume__restore(self,
- n='test_basic_publish'):
- c = memory_client().channel()
-
- c.exchange_declare(n)
- c.queue_declare(n)
- c.queue_bind(n, n, n)
- c.queue_declare(n + '2')
- c.queue_bind(n + '2', n, n)
-
- m = c.prepare_message('nthex quick brown fox...')
- c.basic_publish(m, n, n)
-
- r1 = c.message_to_python(c.basic_get(n))
- self.assertTrue(r1)
- self.assertEqual(r1.body,
- 'nthex quick brown fox...'.encode('utf-8'))
- self.assertIsNone(c.basic_get(n))
-
- consumer_tag = uuid()
-
- c.basic_consume(n + '2', False,
- consumer_tag=consumer_tag, callback=lambda *a: None)
- self.assertIn(n + '2', c._active_queues)
- r2, _ = c.drain_events()
- r2 = c.message_to_python(r2)
- self.assertEqual(r2.body,
- 'nthex quick brown fox...'.encode('utf-8'))
- self.assertEqual(r2.delivery_info['exchange'], n)
- self.assertEqual(r2.delivery_info['routing_key'], n)
- with self.assertRaises(virtual.Empty):
- c.drain_events()
- c.basic_cancel(consumer_tag)
-
- c._restore(r2)
- r3 = c.message_to_python(c.basic_get(n))
- self.assertTrue(r3)
- self.assertEqual(r3.body, 'nthex quick brown fox...'.encode('utf-8'))
- self.assertIsNone(c.basic_get(n))
-
- def test_basic_ack(self):
-
- class MockQoS(virtual.QoS):
- was_acked = False
-
- def ack(self, delivery_tag):
- self.was_acked = True
-
- self.channel._qos = MockQoS(self.channel)
- self.channel.basic_ack('foo')
- self.assertTrue(self.channel._qos.was_acked)
-
- def test_basic_recover__requeue(self):
-
- class MockQoS(virtual.QoS):
- was_restored = False
-
- def restore_unacked(self):
- self.was_restored = True
-
- self.channel._qos = MockQoS(self.channel)
- self.channel.basic_recover(requeue=True)
- self.assertTrue(self.channel._qos.was_restored)
-
- def test_restore_unacked_raises_BaseException(self):
- q = self.channel.qos
- q._flush = Mock()
- q._delivered = {1: 1}
-
- q.channel._restore = Mock()
- q.channel._restore.side_effect = SystemExit
-
- errors = q.restore_unacked()
- self.assertIsInstance(errors[0][0], SystemExit)
- self.assertEqual(errors[0][1], 1)
- self.assertFalse(q._delivered)
-
- @patch('kombu.transport.virtual.emergency_dump_state')
- @patch('kombu.transport.virtual.say')
- def test_restore_unacked_once_when_unrestored(self, say,
- emergency_dump_state):
- q = self.channel.qos
- q._flush = Mock()
-
- class State(dict):
- restored = False
-
- q._delivered = State({1: 1})
- ru = q.restore_unacked = Mock()
- exc = None
- try:
- raise KeyError()
- except KeyError as exc_:
- exc = exc_
- ru.return_value = [(exc, 1)]
-
- self.channel.do_restore = True
- q.restore_unacked_once()
- self.assertTrue(say.called)
- self.assertTrue(emergency_dump_state.called)
-
- def test_basic_recover(self):
- with self.assertRaises(NotImplementedError):
- self.channel.basic_recover(requeue=False)
-
- def test_basic_reject(self):
-
- class MockQoS(virtual.QoS):
- was_rejected = False
-
- def reject(self, delivery_tag, requeue=False):
- self.was_rejected = True
-
- self.channel._qos = MockQoS(self.channel)
- self.channel.basic_reject('foo')
- self.assertTrue(self.channel._qos.was_rejected)
-
- def test_basic_qos(self):
- self.channel.basic_qos(prefetch_count=128)
- self.assertEqual(self.channel._qos.prefetch_count, 128)
-
- def test_lookup__undeliverable(self, n='test_lookup__undeliverable'):
- warnings.resetwarnings()
- with warnings.catch_warnings(record=True) as log:
- self.assertListEqual(
- self.channel._lookup(n, n, 'ae.undeliver'),
- ['ae.undeliver'],
- )
- self.assertTrue(log)
- self.assertIn('could not be delivered', log[0].message.args[0])
-
- def test_context(self):
- x = self.channel.__enter__()
- self.assertIs(x, self.channel)
- x.__exit__()
- self.assertTrue(x.closed)
-
- def test_cycle_property(self):
- self.assertTrue(self.channel.cycle)
-
- def test_flow(self):
- with self.assertRaises(NotImplementedError):
- self.channel.flow(False)
-
- def test_close_when_no_connection(self):
- self.channel.connection = None
- self.channel.close()
- self.assertTrue(self.channel.closed)
-
- def test_drain_events_has_get_many(self):
- c = self.channel
- c._get_many = Mock()
- c._poll = Mock()
- c._consumers = [1]
- c._qos = Mock()
- c._qos.can_consume.return_value = True
-
- c.drain_events(timeout=10.0)
- c._get_many.assert_called_with(c._active_queues, timeout=10.0)
-
- def test_get_exchanges(self):
- self.channel.exchange_declare(exchange='foo')
- self.assertTrue(self.channel.get_exchanges())
-
- def test_basic_cancel_not_in_active_queues(self):
- c = self.channel
- c._consumers.add('x')
- c._tag_to_queue['x'] = 'foo'
- c._active_queues = Mock()
- c._active_queues.remove.side_effect = ValueError()
-
- c.basic_cancel('x')
- c._active_queues.remove.assert_called_with('foo')
-
- def test_basic_cancel_unknown_ctag(self):
- self.assertIsNone(self.channel.basic_cancel('unknown-tag'))
-
- def test_list_bindings(self):
- c = self.channel
- c.exchange_declare(exchange='foo')
- c.queue_declare(queue='q')
- c.queue_bind(queue='q', exchange='foo', routing_key='rk')
-
- self.assertIn(('q', 'foo', 'rk'), list(c.list_bindings()))
-
- def test_after_reply_message_received(self):
- c = self.channel
- c.queue_delete = Mock()
- c.after_reply_message_received('foo')
- c.queue_delete.assert_called_with('foo')
-
- def test_queue_delete_unknown_queue(self):
- self.assertIsNone(self.channel.queue_delete('xiwjqjwel'))
-
- def test_queue_declare_passive(self):
- has_queue = self.channel._has_queue = Mock()
- has_queue.return_value = False
- with self.assertRaises(ChannelError):
- self.channel.queue_declare(queue='21wisdjwqe', passive=True)
-
-
- class test_Transport(Case):
-
- def setUp(self):
- self.transport = client().transport
-
- def test_custom_polling_interval(self):
- x = client(transport_options=dict(polling_interval=32.3))
- self.assertEqual(x.transport.polling_interval, 32.3)
-
- def test_close_connection(self):
- c1 = self.transport.create_channel(self.transport)
- c2 = self.transport.create_channel(self.transport)
- self.assertEqual(len(self.transport.channels), 2)
- self.transport.close_connection(self.transport)
- self.assertFalse(self.transport.channels)
- del(c1) # so pyflakes doesn't complain
- del(c2)
-
- def test_drain_channel(self):
- channel = self.transport.create_channel(self.transport)
- with self.assertRaises(virtual.Empty):
- self.transport._drain_channel(channel)
|