/usr/lib/python3/dist-packages/kombu/tests/utils/test_amq_manager.py is in python3-kombu 3.0.33-1ubuntu2.
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 | from __future__ import absolute_import
from kombu import Connection
from kombu.tests.case import Case, mask_modules, module_exists, patch
class test_get_manager(Case):
@mask_modules('pyrabbit')
def test_without_pyrabbit(self):
with self.assertRaises(ImportError):
Connection('amqp://').get_manager()
@module_exists('pyrabbit')
def test_with_pyrabbit(self):
with patch('pyrabbit.Client', create=True) as Client:
manager = Connection('amqp://').get_manager()
self.assertIsNotNone(manager)
Client.assert_called_with(
'localhost:15672', 'guest', 'guest',
)
@module_exists('pyrabbit')
def test_transport_options(self):
with patch('pyrabbit.Client', create=True) as Client:
manager = Connection('amqp://', transport_options={
'manager_hostname': 'admin.mq.vandelay.com',
'manager_port': 808,
'manager_userid': 'george',
'manager_password': 'bosco',
}).get_manager()
self.assertIsNotNone(manager)
Client.assert_called_with(
'admin.mq.vandelay.com:808', 'george', 'bosco',
)
|