Merge "Add ability to warn on use of non-official types"

os_service_types/exc.py

@@ -0,0 +1,33 @@
+# Copyright 2017 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.
+__all__ = ['warn', 'AliasUsageWarning']
+
+import textwrap
+import warnings
+
+
+def warn(warning, **kwargs):
+    """Emit a warning that has builtin message text."""
+    message = textwrap.fill(textwrap.dedent(warning.details.format(**kwargs)))
+    warnings.warn(message, category=warning)
+
+
+class AliasUsageWarning(Warning):
+    """Use of historical service-type aliases is discouraged."""
+
+    details = """
+    Requested service_type {given} is an old alias. Please update your
+    code to reference the official service_type {official}.
+    """

os_service_types/service_types.py

@@ -17,6 +17,7 @@ __all__ = ['ServiceTypes']
 import copy
 
 import os_service_types.data
+from os_service_types import exc
 
 BUILTIN_DATA = os_service_types.data.read_data('service-types.json')
 SERVICE_TYPES_URL = "https://service-types.openstack.org/service-types.json"
@@ -39,16 +40,20 @@ class ServiceTypes(object):
         remotely the builtin data will be returned as a fallback. only_remote
         will cause remote failures to raise an error instead of falling back.
         Optional, defaults to False.
+    :param bool warn: Emit warnings when a non-official service_type is
+        provided. This provides an easy way for consuming applications to
+        warn users when they are using old types.
     :raises ValueError: If session is None and only_remote is True
     :raises IOError: If session is given and only_remote is True and there is
         an error fetching remote data.
     """
 
-    def __init__(self, session=None, only_remote=False):
+    def __init__(self, session=None, only_remote=False, warn=False):
         if not session and only_remote:
             raise ValueError(
                 "only_remote was requested but no Session was provided.")
         self._service_types_data = BUILTIN_DATA
+        self._warn = warn
         if session:
             try:
                 response = session.get(SERVICE_TYPES_URL)
@@ -194,7 +199,11 @@ class ServiceTypes(object):
         """
         if self.is_official(service_type):
             return service_type
-        return self._service_types_data['reverse'].get(service_type)
+        official = self._service_types_data['reverse'].get(service_type)
+        if self._warn:
+            exc.warn(
+                exc.AliasUsageWarning, given=service_type, official=official)
+        return official
 
     def get_all_types(self, service_type):
         """Get a list of official types and all known aliases.

os_service_types/tests/test_warn.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+
+# 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.
+
+"""
+test_warn
+---------
+
+Tests for warnings
+"""
+import warnings
+
+import os_service_types
+from os_service_types import exc
+from os_service_types.tests import base
+
+
+class TestWarnOn(base.TestCase):
+
+    def setUp(self):
+        super(TestWarnOn, self).setUp()
+        # Cause all warnings to always be triggered.
+        warnings.simplefilter("always")
+        self.service_types = os_service_types.ServiceTypes(warn=True)
+
+    def test_warning_emitted_on_alias(self):
+        with warnings.catch_warnings(record=True) as w:
+            self.service_types.get_service_type('volumev2')
+            self.assertEqual(1, len(w))
+            self.assertTrue(issubclass(w[-1].category, exc.AliasUsageWarning))
+
+    def test_warning_not_emitted_on_official(self):
+        with warnings.catch_warnings(record=True) as w:
+            self.service_types.get_service_type('block-storage')
+            self.assertEqual(0, len(w))
+
+
+class TestWarnOff(base.TestCase):
+
+    def setUp(self):
+        super(TestWarnOff, self).setUp()
+        # Cause all warnings to always be triggered.
+        warnings.simplefilter("always")
+        self.service_types = os_service_types.ServiceTypes()
+
+    def test_warning_not_emitted_on_alias(self):
+        with warnings.catch_warnings(record=True) as w:
+            self.service_types.get_service_type('volumev2')
+            self.assertEqual(0, len(w))
+
+    def test_warning_not_emitted_on_official(self):
+        with warnings.catch_warnings(record=True) as w:
+            self.service_types.get_service_type('block-storage')
+            self.assertEqual(0, len(w))