/usr/lib/python3/dist-packages/kombu/tests/transport/test_transport.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 37 38 39 40 41 42 43 44 | from __future__ import absolute_import
from kombu import transport
from kombu.tests.case import Case, Mock, patch
class test_supports_librabbitmq(Case):
def test_eventlet(self):
with patch('kombu.transport._detect_environment') as de:
de.return_value = 'eventlet'
self.assertFalse(transport.supports_librabbitmq())
class test_transport(Case):
def test_resolve_transport(self):
from kombu.transport.memory import Transport
self.assertIs(transport.resolve_transport(
'kombu.transport.memory:Transport'),
Transport)
self.assertIs(transport.resolve_transport(Transport), Transport)
def test_resolve_transport_alias_callable(self):
m = transport.TRANSPORT_ALIASES['George'] = Mock(name='lazyalias')
try:
transport.resolve_transport('George')
m.assert_called_with()
finally:
transport.TRANSPORT_ALIASES.pop('George')
def test_resolve_transport_alias(self):
self.assertTrue(transport.resolve_transport('pyamqp'))
class test_transport_ghettoq(Case):
@patch('warnings.warn')
def test_compat(self, warn):
x = transport._ghettoq('Redis', 'redis', 'redis')
self.assertEqual(x(), 'kombu.transport.redis.Transport')
self.assertTrue(warn.called)
|