openstack/swift
Code Issues Proposed changes

func tests: Rename storage_url to storage_path

Browse Source 
... so we can use storage_url for actual, complete URLs.
Also, expose storage_scheme for easier comparisons.

Change-Id: Iffcfc327bc674681ac51fe75c1d9911fa3a60d54
This commit is contained in:
Tim Burke 2018-06-12 11:55:31 -07:00
parent 6274ad49cb
commit b39a88ca65
4 changed files with 69 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
test/functional/swift_test_client.py
View File

@ -124,10 +124,42 @@ class Connection(object):
self.password = config['password']
self.storage_netloc = None
self.storage_url = None
self.storage_path = None
self.conn_class = None
@property
def storage_url(self):
return '%s://%s/%s' % (self.storage_scheme, self.storage_netloc,
self.storage_path)
@storage_url.setter
def storage_url(self, value):
url = urllib.parse.urlparse(value)
if url.scheme == 'http':
self.conn_class = http_client.HTTPConnection
elif url.scheme == 'https':
self.conn_class = http_client.HTTPSConnection
else:
raise ValueError('unexpected protocol %s' % (url.scheme))
self.storage_netloc = url.netloc
# Make sure storage_path is a string and not unicode, since
# keystoneclient (called by swiftclient) returns them in
# unicode and this would cause troubles when doing
# no_safe_quote query.
x = url.path.split('/')
self.storage_path = str('/%s/%s' % (x[1], x[2]))
self.account_name = str(x[2])
@property
def storage_scheme(self):
if self.conn_class is None:
return None
if issubclass(self.conn_class, http_client.HTTPSConnection):
return 'https'
return 'http'
def get_account(self):
return Account(self, self.account)
@ -164,23 +196,7 @@ class Connection(object):
if not (storage_url and storage_token):
raise AuthenticationFailed()
url = urllib.parse.urlparse(storage_url)
if url.scheme == 'http':
self.conn_class = http_client.HTTPConnection
elif url.scheme == 'https':
self.conn_class = http_client.HTTPSConnection
else:
raise ValueError('unexpected protocol %s' % (url.scheme))
self.storage_netloc = url.netloc
# Make sure storage_url is a string and not unicode, since
# keystoneclient (called by swiftclient) returns them in
# unicode and this would cause troubles when doing
# no_safe_quote query.
x = url.path.split('/')
self.storage_url = str('/%s/%s' % (x[1], x[2]))
self.account_name = str(x[2])
self.storage_url = storage_url
self.auth_user = auth_user
# With v2 keystone, storage_token is unicode.
# We want it to be string otherwise this would cause
@ -190,7 +206,7 @@ class Connection(object):
self.user_acl = '%s:%s' % (self.account, self.username)
self.http_connect()
return self.storage_url, self.storage_token
return self.storage_path, self.storage_token
def cluster_info(self):
"""
@ -205,7 +221,7 @@ class Connection(object):
return json.loads(self.response.read())
def http_connect(self):
if issubclass(self.conn_class, http_client.HTTPSConnection) and \
if self.storage_scheme == 'https' and \
self.insecure and sys.version_info >= (2, 7, 9):
import ssl
self.connection = self.conn_class(
@ -221,16 +237,16 @@ class Connection(object):
cfg = {}
if cfg.get('version_only_path'):
return '/' + self.storage_url.split('/')[1]
return '/' + self.storage_path.split('/')[1]
if path:
quote = urllib.parse.quote
if cfg.get('no_quote') or cfg.get('no_path_quote'):
quote = lambda x: x
return '%s/%s' % (self.storage_url,
return '%s/%s' % (self.storage_path,
'/'.join([quote(i) for i in path]))
else:
return self.storage_url
return self.storage_path
def make_headers(self, hdrs, cfg=None):
if cfg is None:

15
test/functional/test_domain_remap.py
View File

@ -79,17 +79,22 @@ class TestDomainRemap(Base):
raise SkipTest('Domain Remap storage_domain not configured in %s' %
tf.config['__file__'])
_, _, acct = self.env.account.conn.storage_url.split('/')
storage_domain = tf.config.get('storage_domain')
self.acct_domain_dash = '%s.%s' % (acct, storage_domain)
self.acct_domain_dash = '%s.%s' % (self.env.account.conn.account_name,
storage_domain)
self.acct_domain_underscore = '%s.%s' % (
acct.replace('_', '-'), storage_domain)
self.env.account.conn.account_name.replace('_', '-'),
storage_domain)
self.cont_domain_dash = '%s.%s.%s' % (
self.env.container.name, acct, storage_domain)
self.env.container.name,
self.env.account.conn.account_name,
storage_domain)
self.cont_domain_underscore = '%s.%s.%s' % (
self.env.container.name, acct.replace('_', '-'), storage_domain)
self.env.container.name,
self.env.account.conn.account_name.replace('_', '-'),
storage_domain)
def test_GET_remapped_account(self):
for domain in (self.acct_domain_dash, self.acct_domain_underscore):

28
test/functional/test_staticweb.py
View File

@ -113,16 +113,16 @@ class TestStaticWeb(Base):
def domain_remap_acct(self):
# the storage_domain option is test.conf must be set to one of the
# domain_remap middleware storage_domain values
_, _, acct = self.env.account.conn.storage_url.split('/')
return '.'.join((acct, tf.config.get('storage_domain')))
return '.'.join((self.env.account.conn.account_name,
tf.config.get('storage_domain')))
@property
def domain_remap_cont(self):
# the storage_domain option is test.conf must be set to one of the
# domain_remap middleware storage_domain values
_, _, acct = self.env.account.conn.storage_url.split('/')
return '.'.join(
(self.env.container.name, acct, tf.config.get('storage_domain')))
(self.env.container.name, self.env.account.conn.account_name,
tf.config.get('storage_domain')))
def _set_staticweb_headers(self, index=False, listings=False,
listings_css=False, error=False):
@ -165,11 +165,11 @@ class TestStaticWeb(Base):
def _test_redirect_slash_direct(self, anonymous):
host = self.env.account.conn.storage_netloc
path = '%s/%s' % (self.env.account.conn.storage_url,
path = '%s/%s' % (self.env.account.conn.storage_path,
self.env.container.name)
self._test_redirect_with_slash(host, path, anonymous=anonymous)
path = '%s/%s/%s' % (self.env.account.conn.storage_url,
path = '%s/%s/%s' % (self.env.account.conn.storage_path,
self.env.container.name,
self.env.objects['dir/'].name)
self._test_redirect_with_slash(host, path, anonymous=anonymous)
@ -241,7 +241,7 @@ class TestStaticWeb(Base):
def _test_listing_direct(self, anonymous, listings_css):
objects = self.env.objects
host = self.env.account.conn.storage_netloc
path = '%s/%s/' % (self.env.account.conn.storage_url,
path = '%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name)
css = objects['listings_css'].name if listings_css else None
self._test_listing(host, path, anonymous=True, css=css,
@ -249,7 +249,7 @@ class TestStaticWeb(Base):
objects['dir/'].name + '/'],
notins=[objects['dir/obj'].name])
path = '%s/%s/%s/' % (self.env.account.conn.storage_url,
path = '%s/%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name,
objects['dir/'].name)
css = '../%s' % objects['listings_css'].name if listings_css else None
@ -278,7 +278,7 @@ class TestStaticWeb(Base):
host = self.domain_remap_acct
path = '/%s/' % self.env.container.name
css = objects['listings_css'].name if listings_css else None
title = '%s/%s/' % (self.env.account.conn.storage_url,
title = '%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name)
self._test_listing(host, path, title=title, anonymous=anonymous,
css=css,
@ -288,7 +288,7 @@ class TestStaticWeb(Base):
path = '/%s/%s/' % (self.env.container.name, objects['dir/'].name)
css = '../%s' % objects['listings_css'].name if listings_css else None
title = '%s/%s/%s/' % (self.env.account.conn.storage_url,
title = '%s/%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name,
objects['dir/'])
self._test_listing(host, path, title=title, anonymous=anonymous,
@ -317,7 +317,7 @@ class TestStaticWeb(Base):
host = self.domain_remap_cont
path = '/'
css = objects['listings_css'].name if listings_css else None
title = '%s/%s/' % (self.env.account.conn.storage_url,
title = '%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name)
self._test_listing(host, path, title=title, anonymous=anonymous,
css=css,
@ -327,7 +327,7 @@ class TestStaticWeb(Base):
path = '/%s/' % objects['dir/'].name
css = '../%s' % objects['listings_css'].name if listings_css else None
title = '%s/%s/%s/' % (self.env.account.conn.storage_url,
title = '%s/%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name,
objects['dir/'])
self._test_listing(host, path, title=title, anonymous=anonymous,
@ -366,11 +366,11 @@ class TestStaticWeb(Base):
def _test_index_direct(self, anonymous):
objects = self.env.objects
host = self.env.account.conn.storage_netloc
path = '%s/%s/' % (self.env.account.conn.storage_url,
path = '%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name)
self._test_index(host, path, anonymous=anonymous)
path = '%s/%s/%s/' % (self.env.account.conn.storage_url,
path = '%s/%s/%s/' % (self.env.account.conn.storage_path,
self.env.container.name,
objects['dir/'].name)
self._test_index(host, path, anonymous=anonymous, expected_status=404)

10
test/functional/tests.py
View File

@ -178,16 +178,16 @@ class TestAccount(Base):
self.assert_body('Bad URL')
def testInvalidPath(self):
was_url = self.env.account.conn.storage_url
was_path = self.env.account.conn.storage_path
if (normalized_urls):
self.env.account.conn.storage_url = '/'
self.env.account.conn.storage_path = '/'
else:
self.env.account.conn.storage_url = "/%s" % was_url
self.env.account.conn.make_request('GET')
self.env.account.conn.storage_path = "/%s" % was_path
try:
self.env.account.conn.make_request('GET')
self.assert_status(404)
finally:
self.env.account.conn.storage_url = was_url
self.env.account.conn.storage_path = was_path
def testPUTError(self):
if load_constraint('allow_account_management'):