Merge "Add hacking check for str and unicode in exceptions"

HACKING.rst

@@ -11,6 +11,7 @@ Cinder Specific Commandments
 - [N319] Validate that debug level logs are not translated
 - [N322] Ensure default arguments are not mutable.
 - [N323] Add check for explicit import of _() to ensure proper translation.
+- [N325] str() and unicode() cannot be used on an exception.  Remove or use six.text_type().
 - [N327] assert_called_once is not a valid Mock method.
 - [N328] LOG.info messages require translations `_LI()`.
 - [N329] LOG.exception and LOG.error messages require translations `_LE()`.

cinder/hacking/checks.py

@@ -12,6 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import ast
 import re
 
 """
@@ -57,6 +58,52 @@ log_translation_LW = re.compile(
     r"(.)*LOG\.(warning|warn)\(\s*(_\(|'|\")")
 
 
+class BaseASTChecker(ast.NodeVisitor):
+    """Provides a simple framework for writing AST-based checks.
+
+    Subclasses should implement visit_* methods like any other AST visitor
+    implementation. When they detect an error for a particular node the
+    method should call ``self.add_error(offending_node)``. Details about
+    where in the code the error occurred will be pulled from the node
+    object.
+
+    Subclasses should also provide a class variable named CHECK_DESC to
+    be used for the human readable error message.
+
+    """
+
+    def __init__(self, tree, filename):
+        """This object is created automatically by pep8.
+
+        :param tree: an AST tree
+        :param filename: name of the file being analyzed
+                         (ignored by our checks)
+        """
+        self._tree = tree
+        self._errors = []
+
+    def run(self):
+        """Called automatically by pep8."""
+        self.visit(self._tree)
+        return self._errors
+
+    def add_error(self, node, message=None):
+        """Add an error caused by a node to the list of errors for pep8."""
+
+        # Need to disable pylint check here as it doesn't catch CHECK_DESC
+        # being defined in the subclasses.
+        message = message or self.CHECK_DESC  # pylint: disable=E1101
+        error = (node.lineno, node.col_offset, message, self.__class__)
+        self._errors.append(error)
+
+    def _check_call_names(self, call_node, names):
+        if isinstance(call_node, ast.Call):
+            if isinstance(call_node.func, ast.Name):
+                if call_node.func.id in names:
+                    return True
+        return False
+
+
 def no_vi_headers(physical_line, line_number, lines):
     """Check for vi editor configuration in source files.
 
@@ -115,6 +162,43 @@ def check_explicit_underscore_import(logical_line, filename):
         yield(0, "N323: Found use of _() without explicit import of _ !")
 
 
+class CheckForStrUnicodeExc(BaseASTChecker):
+    """Checks for the use of str() or unicode() on an exception.
+
+    This currently only handles the case where str() or unicode()
+    is used in the scope of an exception handler.  If the exception
+    is passed into a function, returned from an assertRaises, or
+    used on an exception created in the same scope, this does not
+    catch it.
+    """
+
+    CHECK_DESC = ('N325 str() and unicode() cannot be used on an '
+                  'exception.  Remove or use six.text_type()')
+
+    def __init__(self, tree, filename):
+        super(CheckForStrUnicodeExc, self).__init__(tree, filename)
+        self.name = []
+        self.already_checked = []
+
+    def visit_TryExcept(self, node):
+        for handler in node.handlers:
+            if handler.name:
+                self.name.append(handler.name.id)
+                super(CheckForStrUnicodeExc, self).generic_visit(node)
+                self.name = self.name[:-1]
+            else:
+                super(CheckForStrUnicodeExc, self).generic_visit(node)
+
+    def visit_Call(self, node):
+        if self._check_call_names(node, ['str', 'unicode']):
+            if node not in self.already_checked:
+                self.already_checked.append(node)
+                if isinstance(node.args[0], ast.Name):
+                    if node.args[0].id in self.name:
+                        self.add_error(node.args[0])
+        super(CheckForStrUnicodeExc, self).generic_visit(node)
+
+
 def check_assert_called_once(logical_line, filename):
     msg = ("N327: assert_called_once is a no-op. please use assert_called_"
            "once_with to test with explicit parameters or an assertEqual with"
@@ -231,6 +315,7 @@ def factory(register):
     register(no_translate_debug_logs)
     register(no_mutable_default_args)
     register(check_explicit_underscore_import)
+    register(CheckForStrUnicodeExc)
     register(check_assert_called_once)
     register(check_oslo_namespace_imports)
     register(check_datetime_now)

cinder/tests/unit/api/v2/test_limits.py

@@ -366,7 +366,7 @@ class ParseLimitsTest(BaseLimitTestSuite):
                                             '(POST, /bar*, /bar.*, 5, second);'
                                             '(Say, /derp*, /derp.*, 1, day)')
         except ValueError as e:
-            assert False, str(e)
+            assert False, six.text_type(e)
 
         # Make sure the number of returned limits are correct
         self.assertEqual(len(l), 4)

cinder/tests/unit/test_glusterfs.py

@@ -16,6 +16,7 @@
 
 import errno
 import os
+import six
 import tempfile
 import time
 import traceback
@@ -114,7 +115,7 @@ class GlusterFsDriverTestCase(test.TestCase):
             self.assertEqual(excClass, type(exc),
                              'Wrong exception caught: %s Stacktrace: %s' %
                              (exc, traceback.print_exc()))
-            self.assertIn(msg, str(exc))
+            self.assertIn(msg, six.text_type(exc))
 
         if not caught:
             self.fail('Expected raised exception but nothing caught.')

cinder/tests/unit/test_hacking.py

@@ -12,6 +12,11 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import textwrap
+
+import mock
+import pep8
+
 from cinder.hacking import checks
 from cinder import test
 
@@ -114,6 +119,94 @@ class HackingTestCase(test.TestCase):
             "LOG.info('My info message')",
             "cinder.tests.unit/other_files4.py"))))
 
+    # We are patching pep8 so that only the check under test is actually
+    # installed.
+    @mock.patch('pep8._checks',
+                {'physical_line': {}, 'logical_line': {}, 'tree': {}})
+    def _run_check(self, code, checker, filename=None):
+        pep8.register_check(checker)
+
+        lines = textwrap.dedent(code).strip().splitlines(True)
+
+        checker = pep8.Checker(filename=filename, lines=lines)
+        checker.check_all()
+        checker.report._deferred_print.sort()
+        return checker.report._deferred_print
+
+    def _assert_has_errors(self, code, checker, expected_errors=None,
+                           filename=None):
+        actual_errors = [e[:3] for e in
+                         self._run_check(code, checker, filename)]
+        self.assertEqual(expected_errors or [], actual_errors)
+
+    def _assert_has_no_errors(self, code, checker, filename=None):
+        self._assert_has_errors(code, checker, filename=filename)
+
+    def test_str_unicode_exception(self):
+
+        checker = checks.CheckForStrUnicodeExc
+        code = """
+               def f(a, b):
+                   try:
+                       p = str(a) + str(b)
+                   except ValueError as e:
+                       p = str(e)
+                   return p
+               """
+        errors = [(5, 16, 'N325')]
+        self._assert_has_errors(code, checker, expected_errors=errors)
+
+        code = """
+               def f(a, b):
+                   try:
+                       p = unicode(a) + str(b)
+                   except ValueError as e:
+                       p = e
+                   return p
+               """
+        self._assert_has_no_errors(code, checker)
+
+        code = """
+               def f(a, b):
+                   try:
+                       p = str(a) + str(b)
+                   except ValueError as e:
+                       p = unicode(e)
+                   return p
+               """
+        errors = [(5, 20, 'N325')]
+        self._assert_has_errors(code, checker, expected_errors=errors)
+
+        code = """
+               def f(a, b):
+                   try:
+                       p = str(a) + str(b)
+                   except ValueError as e:
+                       try:
+                           p  = unicode(a) + unicode(b)
+                       except ValueError as ve:
+                           p = str(e) + str(ve)
+                       p = e
+                   return p
+               """
+        errors = [(8, 20, 'N325'), (8, 29, 'N325')]
+        self._assert_has_errors(code, checker, expected_errors=errors)
+
+        code = """
+               def f(a, b):
+                   try:
+                       p = str(a) + str(b)
+                   except ValueError as e:
+                       try:
+                           p  = unicode(a) + unicode(b)
+                       except ValueError as ve:
+                           p = str(e) + unicode(ve)
+                       p = str(e)
+                   return p
+               """
+        errors = [(8, 20, 'N325'), (8, 33, 'N325'), (9, 16, 'N325')]
+        self._assert_has_errors(code, checker, expected_errors=errors)
+
     def test_check_no_log_audit(self):
         self.assertEqual(1, len(list(checks.check_no_log_audit(
             "LOG.audit('My test audit log')"))))

cinder/tests/unit/test_quobyte.py

@@ -17,6 +17,7 @@
 
 import errno
 import os
+import six
 import StringIO
 import traceback
 
@@ -107,7 +108,7 @@ class QuobyteDriverTestCase(test.TestCase):
             self.assertEqual(excClass, type(exc),
                              'Wrong exception caught: %s Stacktrace: %s' %
                              (exc, traceback.print_exc()))
-            self.assertIn(msg, str(exc))
+            self.assertIn(msg, six.text_type(exc))
 
         if not caught:
             self.fail('Expected raised exception but nothing caught.')

cinder/volume/drivers/nimble.py

@@ -21,6 +21,7 @@ This driver supports Nimble Storage controller CS-Series.
 import functools
 import random
 import re
+import six
 import string
 import urllib2
 
@@ -389,7 +390,8 @@ def _connection_checker(func):
             try:
                 return func(self, *args, **kwargs)
             except NimbleAPIException as e:
-                if attempts < 1 and (re.search('SM-eaccess', str(e))):
+                if attempts < 1 and (re.search('SM-eaccess',
+                                     six.text_type(e))):
                     LOG.info(_LI('Session might have expired.'
                                  ' Trying to relogin'))
                     self.login()

cinder/volume/drivers/san/hp/hp_3par_common.py

@@ -40,6 +40,7 @@ import json
 import math
 import pprint
 import re
+import six
 import uuid
 
 from oslo_utils import importutils
@@ -1842,7 +1843,7 @@ class HP3PARCommon(object):
                      'userCPG': new_cpg,
                      'conversionOperation': cop})
             except hpexceptions.HTTPBadRequest as ex:
-                if ex.get_code() == 40 and "keepVV" in str(ex):
+                if ex.get_code() == 40 and "keepVV" in six.text_type(ex):
                     # Cannot retype with snapshots because we don't want to
                     # use keepVV and have straggling volumes.  Log additional
                     # info and then raise.