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)