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