use only exceptions for uri validations

Currently we use asserts for uri validation, which is not good
practice because assert will be deleted from optimized code.
We must use exceptions in such cases.

Co-authored-by: kairat_kushaev <kkushaev@mirantis.com>
Co-authored-by: Brian Rosmaita <rosmaita.fossdev@gmail.com>

Change-Id: I89c5f1b74be89c759d5754d6cab54dc86c946be1

glance_store/_drivers/cinder.py

@@ -379,10 +379,7 @@ class StoreLocation(glance_store.location.StoreLocation):
         return "cinder://%s" % self.volume_id
 
     def parse_uri(self, uri):
-        if not uri.startswith('cinder://'):
-            reason = _("URI must start with 'cinder://'")
-            LOG.info(reason)
-            raise exceptions.BadStoreUri(message=reason)
+        self.validate_schemas(uri, valid_schemas=('cinder://',))
 
         self.scheme = 'cinder'
         self.volume_id = uri[9:]

glance_store/_drivers/filesystem.py

@@ -180,7 +180,7 @@ class StoreLocation(glance_store.location.StoreLocation):
         versions of Python.
         """
         pieces = urllib.parse.urlparse(uri)
-        assert pieces.scheme in ('file', 'filesystem')
+        self.validate_schemas(uri, valid_schemas=('file://', 'filesystem://'))
         self.scheme = pieces.scheme
         path = (pieces.netloc + pieces.path).strip()
         if path == '':

glance_store/_drivers/http.py

@@ -125,7 +125,7 @@ class StoreLocation(glance_store.location.StoreLocation):
         versions of Python.
         """
         pieces = urllib.parse.urlparse(uri)
-        assert pieces.scheme in ('https', 'http')
+        self.validate_schemas(uri, valid_schemas=('https://', 'http://'))
         self.scheme = pieces.scheme
         netloc = pieces.netloc
         path = pieces.path

glance_store/_drivers/rbd.py

@@ -182,12 +182,7 @@ class StoreLocation(location.StoreLocation):
 
     def parse_uri(self, uri):
         prefix = 'rbd://'
-        if not uri.startswith(prefix):
-            reason = _('URI must start with rbd://')
-            msg = _LI("Invalid URI: %s") % reason
-
-            LOG.info(msg)
-            raise exceptions.BadStoreUri(message=reason)
+        self.validate_schemas(uri, valid_schemas=(prefix,))
         # convert to ascii since librbd doesn't handle unicode
         try:
             ascii_uri = str(uri)

glance_store/_drivers/sheepdog.py

@@ -222,9 +222,7 @@ class StoreLocation(glance_store.location.StoreLocation):
 
     def parse_uri(self, uri):
         valid_schema = 'sheepdog://'
-        if not uri.startswith(valid_schema):
-            reason = _("URI must start with '%s'") % valid_schema
-            raise exceptions.BadStoreUri(message=reason)
+        self.validate_schemas(uri, valid_schemas=(valid_schema,))
         pieces = uri[len(valid_schema):].split(':')
         if len(pieces) == 3:
             self.image = pieces[2]

glance_store/_drivers/swift/store.py

@@ -685,8 +685,8 @@ class StoreLocation(location.StoreLocation):
             raise exceptions.BadStoreUri(message=reason)
 
         pieces = urllib.parse.urlparse(uri)
-        assert pieces.scheme in ('swift', 'swift+http', 'swift+https',
-                                 'swift+config')
+        self.validate_schemas(uri, valid_schemas=(
+            'swift://', 'swift+http://', 'swift+https://', 'swift+config://'))
 
         self.scheme = pieces.scheme
         netloc = pieces.netloc

glance_store/_drivers/vmware_datastore.py

@@ -320,11 +320,7 @@ class StoreLocation(location.StoreLocation):
     #    return path.startswith(os.path.join(DS_URL_PREFIX, sdir))
 
     def parse_uri(self, uri):
-        if not uri.startswith('%s://' % STORE_SCHEME):
-            reason = (_("URI %(uri)s must start with %(scheme)s://") %
-                      {'uri': uri, 'scheme': STORE_SCHEME})
-            LOG.info(reason)
-            raise exceptions.BadStoreUri(message=reason)
+        self.validate_schemas(uri, valid_schemas=('%s://' % STORE_SCHEME,))
         (self.scheme, self.server_host,
          path, params, query, fragment) = urllib.parse.urlparse(uri)
         if not query:

glance_store/location.py

@@ -43,6 +43,7 @@ from oslo_config import cfg
 from six.moves import urllib
 
 from glance_store import exceptions
+from glance_store.i18n import _
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -167,3 +168,17 @@ class StoreLocation(object):
         """
         raise NotImplementedError("StoreLocation subclass must implement "
                                   "parse_uri()")
+
+    @staticmethod
+    def validate_schemas(uri, valid_schemas):
+        """check if uri scheme is one of valid_schemas
+        generate exception otherwise
+        """
+        for valid_schema in valid_schemas:
+            if uri.startswith(valid_schema):
+                return
+
+        reason = _("Location URI must start with one of the following "
+                   "schemas: %s") % ', '.join(valid_schemas)
+        LOG.warning(reason)
+        raise exceptions.BadStoreUri(message=reason)

glance_store/tests/unit/test_location.py

@@ -0,0 +1,30 @@
+#    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.
+
+from glance_store import exceptions
+from glance_store import location
+from glance_store.tests import base
+
+
+class TestStoreLocation(base.StoreBaseTest):
+
+    def setUp(self):
+        super(TestStoreLocation, self).setUp()
+
+    def test_scheme_validation(self):
+        valid_schemas = ("file://", "http://")
+        correct_uri = "file://test"
+        location.StoreLocation.validate_schemas(correct_uri, valid_schemas)
+        incorrect_uri = "fake://test"
+        self.assertRaises(exceptions.BadStoreUri,
+                          location.StoreLocation.validate_schemas,
+                          incorrect_uri, valid_schemas)