/usr/lib/python2.7/dist-packages/celery/tests/backends/test_rpc.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | from __future__ import absolute_import
from celery.backends.rpc import RPCBackend
from celery._state import _task_stack
from celery.tests.case import AppCase, Mock, patch
class test_RPCBackend(AppCase):
def setup(self):
self.b = RPCBackend(app=self.app)
def test_oid(self):
oid = self.b.oid
oid2 = self.b.oid
self.assertEqual(oid, oid2)
self.assertEqual(oid, self.app.oid)
def test_interface(self):
self.b.on_reply_declare('task_id')
def test_destination_for(self):
req = Mock(name='request')
req.reply_to = 'reply_to'
req.correlation_id = 'corid'
self.assertTupleEqual(
self.b.destination_for('task_id', req),
('reply_to', 'corid'),
)
task = Mock()
_task_stack.push(task)
try:
task.request.reply_to = 'reply_to'
task.request.correlation_id = 'corid'
self.assertTupleEqual(
self.b.destination_for('task_id', None),
('reply_to', 'corid'),
)
finally:
_task_stack.pop()
with self.assertRaises(RuntimeError):
self.b.destination_for('task_id', None)
def test_binding(self):
queue = self.b.binding
self.assertEqual(queue.name, self.b.oid)
self.assertEqual(queue.exchange, self.b.exchange)
self.assertEqual(queue.routing_key, self.b.oid)
self.assertFalse(queue.durable)
self.assertFalse(queue.auto_delete)
def test_many_bindings(self):
self.assertListEqual(
self.b._many_bindings(['a', 'b']),
[self.b.binding],
)
def test_create_binding(self):
self.assertEqual(self.b._create_binding('id'), self.b.binding)
def test_on_task_call(self):
with patch('celery.backends.rpc.maybe_declare') as md:
with self.app.amqp.producer_pool.acquire() as prod:
self.b.on_task_call(prod, 'task_id'),
md.assert_called_with(
self.b.binding(prod.channel),
retry=True,
)
def test_create_exchange(self):
ex = self.b._create_exchange('name')
self.assertIsInstance(ex, self.b.Exchange)
self.assertEqual(ex.name, '')
|