Refactor extract code to function in auth.py

Code for applying filters to the urls in the KeystoneV*AuthProvider
classes was copy-pasted. Extracting the copied code to a function so
that fixes can be made in one place.

Change-Id: I6b1029d2004fe0208519b506c2751390a6c49569

tempest/lib/auth.py

@@ -31,6 +31,37 @@ ISO8601_INT_SECONDS = '%Y-%m-%dT%H:%M:%SZ'
 LOG = logging.getLogger(__name__)
 
 
+def replace_version(url, new_version):
+    parts = urlparse.urlparse(url)
+    version_path = '/%s' % new_version
+    path = re.sub(r'(^|/)+v\d+(?:\.\d+)?',
+                  version_path,
+                  parts.path,
+                  count=1)
+    url = urlparse.urlunparse((parts.scheme,
+                               parts.netloc,
+                               path or version_path,
+                               parts.params,
+                               parts.query,
+                               parts.fragment))
+    return url
+
+
+def apply_url_filters(url, filters):
+    if filters.get('api_version', None) is not None:
+        url = replace_version(url, filters['api_version'])
+    parts = urlparse.urlparse(url)
+    if filters.get('skip_path', None) is not None and parts.path != '':
+        url = urlparse.urlunparse((parts.scheme,
+                                   parts.netloc,
+                                   '/',
+                                   parts.params,
+                                   parts.query,
+                                   parts.fragment))
+
+    return url
+
+
 @six.add_metaclass(abc.ABCMeta)
 class AuthProvider(object):
     """Provide authentication"""
@@ -322,29 +353,7 @@ class KeystoneV2AuthProvider(KeystoneAuthProvider):
             raise exceptions.EndpointNotFound(
                 "service: %s, region: %s, endpoint_type: %s" %
                 (service, region, endpoint_type))
-
-        parts = urlparse.urlparse(_base_url)
-        if filters.get('api_version', None) is not None:
-            version_path = '/%s' % filters['api_version']
-            path = re.sub(r'(^|/)+v\d+(?:\.\d+)?',
-                          version_path,
-                          parts.path,
-                          count=1)
-            _base_url = urlparse.urlunparse((parts.scheme,
-                                             parts.netloc,
-                                             path or version_path,
-                                             parts.params,
-                                             parts.query,
-                                             parts.fragment))
-        if filters.get('skip_path', None) is not None and parts.path != '':
-            _base_url = urlparse.urlunparse((parts.scheme,
-                                             parts.netloc,
-                                             '/',
-                                             parts.params,
-                                             parts.query,
-                                             parts.fragment))
-
-        return _base_url
+        return apply_url_filters(_base_url, filters)
 
     def is_expired(self, auth_data):
         _, access = auth_data
@@ -455,29 +464,7 @@ class KeystoneV3AuthProvider(KeystoneAuthProvider):
         _base_url = filtered_catalog[0].get('url', None)
         if _base_url is None:
                 raise exceptions.EndpointNotFound(service)
-
-        parts = urlparse.urlparse(_base_url)
-        if filters.get('api_version', None) is not None:
-            version_path = '/%s' % filters['api_version']
-            path = re.sub(r'(^|/)+v\d+(?:\.\d+)?',
-                          version_path,
-                          parts.path,
-                          count=1)
-            _base_url = urlparse.urlunparse((parts.scheme,
-                                             parts.netloc,
-                                             path or version_path,
-                                             parts.params,
-                                             parts.query,
-                                             parts.fragment))
-        if filters.get('skip_path', None) is not None:
-            _base_url = urlparse.urlunparse((parts.scheme,
-                                             parts.netloc,
-                                             '/',
-                                             parts.params,
-                                             parts.query,
-                                             parts.fragment))
-
-        return _base_url
+        return apply_url_filters(_base_url, filters)
 
     def is_expired(self, auth_data):
         _, access = auth_data

tempest/tests/lib/test_auth.py

@@ -570,3 +570,68 @@ class TestKeystoneV3Credentials(base.TestCase):
         attrs = {'tenant_name': 'tenant', 'project_name': 'project'}
         self.assertRaises(
             exceptions.InvalidCredentials, auth.KeystoneV3Credentials, **attrs)
+
+
+class TestReplaceVersion(base.TestCase):
+    def test_version_no_trailing_path(self):
+        self.assertEqual(
+            'http://localhost:35357/v2.0',
+            auth.replace_version('http://localhost:35357/v3', 'v2.0'))
+
+    def test_version_no_trailing_path_solidus(self):
+        self.assertEqual(
+            'http://localhost:35357/v2.0/',
+            auth.replace_version('http://localhost:35357/v3/', 'v2.0'))
+
+    def test_version_trailing_path(self):
+        self.assertEqual(
+            'http://localhost:35357/v2.0/uuid',
+            auth.replace_version('http://localhost:35357/v3/uuid', 'v2.0'))
+
+    def test_version_trailing_path_solidus(self):
+        self.assertEqual(
+            'http://localhost:35357/v2.0/uuid/',
+            auth.replace_version('http://localhost:35357/v3/uuid/', 'v2.0'))
+
+    def test_no_version_base(self):
+        self.assertEqual(
+            'http://localhost:35357/v2.0',
+            auth.replace_version('http://localhost:35357', 'v2.0'))
+
+    def test_no_version_base_solidus(self):
+        # TODO(blk-u): This doesn't look like it works as expected.
+        self.assertEqual(
+            'http://localhost:35357/',
+            auth.replace_version('http://localhost:35357/', 'v2.0'))
+
+    def test_no_version_path(self):
+        # TODO(blk-u): This doesn't look like it works as expected.
+        self.assertEqual(
+            'http://localhost/identity',
+            auth.replace_version('http://localhost/identity', 'v2.0'))
+
+    def test_no_version_path_solidus(self):
+        # TODO(blk-u): This doesn't look like it works as expected.
+        self.assertEqual(
+            'http://localhost/identity/',
+            auth.replace_version('http://localhost/identity/', 'v2.0'))
+
+    def test_path_version(self):
+        self.assertEqual(
+            'http://localhost/identity/v2.0',
+            auth.replace_version('http://localhost/identity/v3', 'v2.0'))
+
+    def test_path_version_solidus(self):
+        self.assertEqual(
+            'http://localhost/identity/v2.0/',
+            auth.replace_version('http://localhost/identity/v3/', 'v2.0'))
+
+    def test_path_version_trailing_path(self):
+        self.assertEqual(
+            'http://localhost/identity/v2.0/uuid',
+            auth.replace_version('http://localhost/identity/v3/uuid', 'v2.0'))
+
+    def test_path_version_trailing_path_solidus(self):
+        self.assertEqual(
+            'http://localhost/identity/v2.0/uuid/',
+            auth.replace_version('http://localhost/identity/v3/uuid/', 'v2.0'))