oslo.messaging/oslo_messaging/tests/test_urls.py
JiaJunsu 2aab5a6370 Add parse.unquote to transport_url
To support special chars in username:password and vhost

Closes-Bug: #1557496

Change-Id: I977bce06c6b1bb71162d49de82a6a3d62df3e6bd
2016-05-07 05:55:12 -07:00

246 lines
9.4 KiB
Python

# Copyright 2013 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import testscenarios
import oslo_messaging
from oslo_messaging.tests import utils as test_utils
load_tests = testscenarios.load_tests_apply_scenarios
class TestParseURL(test_utils.BaseTestCase):
scenarios = [
('transport',
dict(url='foo:', aliases=None,
expect=dict(transport='foo'))),
('transport_aliased',
dict(url='bar:', aliases=dict(bar='foo'),
expect=dict(transport='foo'))),
('virtual_host_slash',
dict(url='foo:////', aliases=None,
expect=dict(transport='foo', virtual_host='/'))),
('virtual_host',
dict(url='foo:///bar', aliases=None,
expect=dict(transport='foo', virtual_host='bar'))),
('host',
dict(url='foo://host/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host'),
]))),
('ipv6_host',
dict(url='foo://[ffff::1]/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='ffff::1'),
]))),
('port',
dict(url='foo://host:1234/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host', port=1234),
]))),
('ipv6_port',
dict(url='foo://[ffff::1]:1234/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='ffff::1', port=1234),
]))),
('username',
dict(url='foo://u@host:1234/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host', port=1234, username='u'),
]))),
('password',
dict(url='foo://u:p@host:1234/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host', port=1234,
username='u', password='p'),
]))),
('creds_no_host',
dict(url='foo://u:p@/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(username='u', password='p'),
]))),
('multi_host',
dict(url='foo://u:p@host1:1234,host2:4321/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host1', port=1234,
username='u', password='p'),
dict(host='host2', port=4321),
]))),
('multi_creds',
dict(url='foo://u1:p1@host1:1234,u2:p2@host2:4321/bar', aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host1', port=1234,
username='u1', password='p1'),
dict(host='host2', port=4321,
username='u2', password='p2'),
]))),
('multi_creds_ipv6',
dict(url='foo://u1:p1@[ffff::1]:1234,u2:p2@[ffff::2]:4321/bar',
aliases=None,
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='ffff::1', port=1234,
username='u1', password='p1'),
dict(host='ffff::2', port=4321,
username='u2', password='p2'),
]))),
('quoting',
dict(url='foo://u%24:p%26@host:1234/%24', aliases=None,
expect=dict(transport='foo',
virtual_host='$',
hosts=[
dict(host='host', port=1234,
username='u$', password='p&'),
]))),
]
def test_parse_url(self):
self.config(rpc_backend=None)
url = oslo_messaging.TransportURL.parse(self.conf, self.url,
self.aliases)
hosts = []
for host in self.expect.get('hosts', []):
hosts.append(oslo_messaging.TransportHost(host.get('host'),
host.get('port'),
host.get('username'),
host.get('password')))
expected = oslo_messaging.TransportURL(self.conf,
self.expect.get('transport'),
self.expect.get('virtual_host'),
hosts)
self.assertEqual(expected, url)
class TestFormatURL(test_utils.BaseTestCase):
scenarios = [
('rpc_backend',
dict(rpc_backend='testbackend',
transport=None,
virtual_host=None,
hosts=[],
aliases=None,
expected='testbackend:///')),
('rpc_backend_aliased',
dict(rpc_backend='testfoo',
transport=None,
virtual_host=None,
hosts=[],
aliases=dict(testfoo='testbackend'),
expected='testbackend:///')),
('transport',
dict(rpc_backend=None,
transport='testtransport',
virtual_host=None,
hosts=[],
aliases=None,
expected='testtransport:///')),
('transport_aliased',
dict(rpc_backend=None,
transport='testfoo',
virtual_host=None,
hosts=[],
aliases=dict(testfoo='testtransport'),
expected='testtransport:///')),
('virtual_host',
dict(rpc_backend=None,
transport='testtransport',
virtual_host='/vhost',
hosts=[],
aliases=None,
expected='testtransport:////vhost')),
('host',
dict(rpc_backend=None,
transport='testtransport',
virtual_host='/',
hosts=[
dict(hostname='host',
port=10,
username='bob',
password='secret'),
],
aliases=None,
expected='testtransport://bob:secret@host:10//')),
('multi_host',
dict(rpc_backend=None,
transport='testtransport',
virtual_host='',
hosts=[
dict(hostname='h1',
port=1000,
username='b1',
password='s1'),
dict(hostname='h2',
port=2000,
username='b2',
password='s2'),
],
aliases=None,
expected='testtransport://b1:s1@h1:1000,b2:s2@h2:2000/')),
('quoting',
dict(rpc_backend=None,
transport='testtransport',
virtual_host='/$',
hosts=[
dict(hostname='host',
port=10,
username='b$',
password='s&'),
],
aliases=None,
expected='testtransport://b%24:s%26@host:10//%24')),
]
def test_parse_url(self):
self.config(rpc_backend=self.rpc_backend)
hosts = []
for host in self.hosts:
hosts.append(oslo_messaging.TransportHost(host.get('hostname'),
host.get('port'),
host.get('username'),
host.get('password')))
url = oslo_messaging.TransportURL(self.conf,
self.transport,
self.virtual_host,
hosts,
self.aliases)
self.assertEqual(self.expected, str(url))