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
|
from __future__ import absolute_import
from nose import SkipTest
from kombu.connection import BrokerConnection
from kombu.tests.utils import TestCase, skip_if_not_module
class MockConnection(dict):
def __setattr__(self, key, value):
self[key] = value
class test_mongodb(TestCase):
@skip_if_not_module('pymongo')
def test_url_parser(self):
from kombu.transport import mongodb
from pymongo.errors import ConfigurationError
raise SkipTest(
'Test is functional: it actually connects to mongod')
class Transport(mongodb.Transport):
Connection = MockConnection
url = 'mongodb://'
c = BrokerConnection(url, transport=Transport).connect()
client = c.channels[0].client
self.assertEquals(client.name, 'kombu_default')
self.assertEquals(client.connection.host, '127.0.0.1')
url = 'mongodb://localhost'
c = BrokerConnection(url, transport=Transport).connect()
client = c.channels[0].client
self.assertEquals(client.name, 'kombu_default')
url = 'mongodb://localhost/dbname'
c = BrokerConnection(url, transport=Transport).connect()
client = c.channels[0].client
self.assertEquals(client.name, 'dbname')
url = 'mongodb://localhost,example.org:29017/dbname'
c = BrokerConnection(url, transport=Transport).connect()
client = c.channels[0].client
nodes = client.connection.nodes
self.assertEquals(len(nodes), 2)
self.assertTrue(('example.org', 29017) in nodes)
self.assertEquals(client.name, 'dbname')
# Passing options breaks kombu's _init_params method
# url = 'mongodb://localhost,localhost2:29017/dbname?safe=true'
# c = BrokerConnection(url, transport=Transport).connect()
# client = c.channels[0].client
url = 'mongodb://localhost:27017,localhost2:29017/dbname'
c = BrokerConnection(url, transport=Transport).connect()
client = c.channels[0].client
url = 'mongodb://username:password@localhost/dbname'
c = BrokerConnection(url, transport=Transport).connect()
# Assuming there's no user 'username' with password 'password'
# configured in mongodb
# Needed, otherwise the error would be rose before
# the assertRaises is called
def get_client():
c.channels[0].client
self.assertRaises(ConfigurationError, get_client)
|
