OpenStack library for messaging
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
oslo.messaging/oslo_messaging/tests/test_urls.py

225 lines
8.5 KiB

# 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:',
expect=dict(transport='foo'))),
('virtual_host_slash',
dict(url='foo:////',
expect=dict(transport='foo', virtual_host='/'))),
('virtual_host',
dict(url='foo:///bar',
expect=dict(transport='foo', virtual_host='bar'))),
('host',
dict(url='foo://host/bar',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host'),
]))),
('ipv6_host',
dict(url='foo://[ffff::1]/bar',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='ffff::1'),
]))),
('port',
dict(url='foo://host:1234/bar',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host', port=1234),
]))),
('ipv6_port',
dict(url='foo://[ffff::1]:1234/bar',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='ffff::1', port=1234),
]))),
('username',
dict(url='foo://u@host:1234/bar',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host', port=1234, username='u'),
]))),
('password',
dict(url='foo://u:p@host:1234/bar',
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',
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',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host1', port=1234,
username='u', password='p'),
dict(host='host2', port=4321),
]))),
('multi_host_partial_creds',
dict(url='foo://u:p@host1,host2/bar',
expect=dict(transport='foo',
virtual_host='bar',
hosts=[
dict(host='host1', username='u', password='p'),
dict(host='host2'),
]))),
('multi_creds',
dict(url='foo://u1:p1@host1:1234,u2:p2@host2:4321/bar',
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',
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',
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)
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=[],
expected='testbackend:///')),
('transport',
dict(rpc_backend=None,
transport='testtransport',
virtual_host=None,
hosts=[],
expected='testtransport:///')),
('virtual_host',
dict(rpc_backend=None,
transport='testtransport',
virtual_host='/vhost',
hosts=[],
expected='testtransport:////vhost')),
('host',
dict(rpc_backend=None,
transport='testtransport',
virtual_host='/',
hosts=[
dict(hostname='host',
port=10,
username='bob',
password='secret'),
],
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'),
],
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&'),
],
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.assertEqual(self.expected, str(url))