diff options
Diffstat (limited to 'tests/test_urls.py')
| -rw-r--r-- | tests/test_urls.py | 236 |
1 files changed, 0 insertions, 236 deletions
diff --git a/tests/test_urls.py b/tests/test_urls.py deleted file mode 100644 index 9562742..0000000 --- a/tests/test_urls.py +++ /dev/null @@ -1,236 +0,0 @@ - -# 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 - -from oslo import 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'), - ]))), - ] - - def test_parse_url(self): - self.config(rpc_backend=None) - - url = messaging.TransportURL.parse(self.conf, self.url, self.aliases) - - hosts = [] - for host in self.expect.get('hosts', []): - hosts.append(messaging.TransportHost(host.get('host'), - host.get('port'), - host.get('username'), - host.get('password'))) - expected = 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(messaging.TransportHost(host.get('hostname'), - host.get('port'), - host.get('username'), - host.get('password'))) - - url = messaging.TransportURL(self.conf, - self.transport, - self.virtual_host, - hosts, - self.aliases) - - self.assertEqual(self.expected, str(url)) |