1
2# Copyright 2013 Red Hat, Inc.
3#
4#    Licensed under the Apache License, Version 2.0 (the "License"); you may
5#    not use this file except in compliance with the License. You may obtain
6#    a copy of the License at
7#
8#         http://www.apache.org/licenses/LICENSE-2.0
9#
10#    Unless required by applicable law or agreed to in writing, software
11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13#    License for the specific language governing permissions and limitations
14#    under the License.
15
16import testscenarios
17
18import oslo_messaging
19from oslo_messaging.tests import utils as test_utils
20
21load_tests = testscenarios.load_tests_apply_scenarios
22
23
24class TestParseURL(test_utils.BaseTestCase):
25
26    scenarios = [
27        ('transport',
28         dict(url='foo:',
29              expect=dict(transport='foo'))),
30        ('virtual_host_slash',
31         dict(url='foo:////',
32              expect=dict(transport='foo', virtual_host='/'))),
33        ('virtual_host',
34         dict(url='foo:///bar',
35              expect=dict(transport='foo', virtual_host='bar'))),
36        ('host',
37         dict(url='foo://host/bar',
38              expect=dict(transport='foo',
39                          virtual_host='bar',
40                          hosts=[
41                              dict(host='host'),
42                          ]))),
43        ('ipv6_host',
44         dict(url='foo://[ffff::1]/bar',
45              expect=dict(transport='foo',
46                          virtual_host='bar',
47                          hosts=[
48                              dict(host='ffff::1'),
49                          ]))),
50        ('port',
51         dict(url='foo://host:1234/bar',
52              expect=dict(transport='foo',
53                          virtual_host='bar',
54                          hosts=[
55                              dict(host='host', port=1234),
56                          ]))),
57        ('ipv6_port',
58         dict(url='foo://[ffff::1]:1234/bar',
59              expect=dict(transport='foo',
60                          virtual_host='bar',
61                          hosts=[
62                              dict(host='ffff::1', port=1234),
63                          ]))),
64        ('username',
65         dict(url='foo://u@host:1234/bar',
66              expect=dict(transport='foo',
67                          virtual_host='bar',
68                          hosts=[
69                              dict(host='host', port=1234, username='u'),
70                          ]))),
71        ('password',
72         dict(url='foo://u:p@host:1234/bar',
73              expect=dict(transport='foo',
74                          virtual_host='bar',
75                          hosts=[
76                              dict(host='host', port=1234,
77                                   username='u', password='p'),
78                          ]))),
79        ('creds_no_host',
80         dict(url='foo://u:p@/bar',
81              expect=dict(transport='foo',
82                          virtual_host='bar',
83                          hosts=[
84                              dict(username='u', password='p'),
85                          ]))),
86        ('multi_host',
87         dict(url='foo://u:p@host1:1234,host2:4321/bar',
88              expect=dict(transport='foo',
89                          virtual_host='bar',
90                          hosts=[
91                              dict(host='host1', port=1234,
92                                   username='u', password='p'),
93                              dict(host='host2', port=4321),
94                          ]))),
95        ('multi_host_partial_creds',
96         dict(url='foo://u:p@host1,host2/bar',
97              expect=dict(transport='foo',
98                          virtual_host='bar',
99                          hosts=[
100                              dict(host='host1', username='u', password='p'),
101                              dict(host='host2'),
102                          ]))),
103        ('multi_creds',
104         dict(url='foo://u1:p1@host1:1234,u2:p2@host2:4321/bar',
105              expect=dict(transport='foo',
106                          virtual_host='bar',
107                          hosts=[
108                              dict(host='host1', port=1234,
109                                   username='u1', password='p1'),
110                              dict(host='host2', port=4321,
111                                   username='u2', password='p2'),
112                          ]))),
113        ('multi_creds_ipv6',
114         dict(url='foo://u1:p1@[ffff::1]:1234,u2:p2@[ffff::2]:4321/bar',
115              expect=dict(transport='foo',
116                          virtual_host='bar',
117                          hosts=[
118                              dict(host='ffff::1', port=1234,
119                                   username='u1', password='p1'),
120                              dict(host='ffff::2', port=4321,
121                                   username='u2', password='p2'),
122                          ]))),
123        ('quoting',
124         dict(url='foo://u%24:p%26@host:1234/%24',
125              expect=dict(transport='foo',
126                          virtual_host='$',
127                          hosts=[
128                              dict(host='host', port=1234,
129                                   username='u$', password='p&'),
130                          ]))),
131    ]
132
133    def test_parse_url(self):
134        url = oslo_messaging.TransportURL.parse(self.conf, self.url)
135
136        hosts = []
137        for host in self.expect.get('hosts', []):
138            hosts.append(oslo_messaging.TransportHost(host.get('host'),
139                                                      host.get('port'),
140                                                      host.get('username'),
141                                                      host.get('password')))
142        expected = oslo_messaging.TransportURL(self.conf,
143                                               self.expect.get('transport'),
144                                               self.expect.get('virtual_host'),
145                                               hosts)
146
147        self.assertEqual(expected, url)
148
149
150class TestFormatURL(test_utils.BaseTestCase):
151
152    scenarios = [
153        ('transport',
154         dict(transport='testtransport',
155              virtual_host=None,
156              hosts=[],
157              expected='testtransport:///')),
158        ('virtual_host',
159         dict(transport='testtransport',
160              virtual_host='/vhost',
161              hosts=[],
162              expected='testtransport:////vhost')),
163        ('host',
164         dict(transport='testtransport',
165              virtual_host='/',
166              hosts=[
167                  dict(hostname='host',
168                       port=10,
169                       username='bob',
170                       password='secret'),
171              ],
172              expected='testtransport://bob:secret@host:10//')),
173        ('multi_host',
174         dict(transport='testtransport',
175              virtual_host='',
176              hosts=[
177                  dict(hostname='h1',
178                       port=1000,
179                       username='b1',
180                       password='s1'),
181                  dict(hostname='h2',
182                       port=2000,
183                       username='b2',
184                       password='s2'),
185              ],
186              expected='testtransport://b1:s1@h1:1000,b2:s2@h2:2000/')),
187        ('quoting',
188         dict(transport='testtransport',
189              virtual_host='/$',
190              hosts=[
191                  dict(hostname='host',
192                       port=10,
193                       username='b$',
194                       password='s&'),
195              ],
196              expected='testtransport://b%24:s%26@host:10//%24')),
197    ]
198
199    def test_parse_url(self):
200        hosts = []
201        for host in self.hosts:
202            hosts.append(oslo_messaging.TransportHost(host.get('hostname'),
203                                                      host.get('port'),
204                                                      host.get('username'),
205                                                      host.get('password')))
206
207        url = oslo_messaging.TransportURL(self.conf, self.transport,
208                                          self.virtual_host, hosts)
209
210        self.assertEqual(self.expected, str(url))
211