- [N333] Ensure that oslo namespaces are used for namespaced libraries.
- [N339] Prevent use of deprecated contextlib.nested.
- [C301] timeutils.utcnow() from oslo_utils should be used instead of datetime.now().
+- [C302] six.text_type should be used instead of unicode
General
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_utils import strutils
+import six
import webob
from cinder.api import extensions
except exception.InvalidVolume as error:
raise webob.exc.HTTPBadRequest(explanation=error.msg)
except ValueError as error:
- raise webob.exc.HTTPBadRequest(explanation=unicode(error))
+ raise webob.exc.HTTPBadRequest(explanation=six.text_type(error))
except messaging.RemoteError as error:
msg = "%(err_type)s: %(err_msg)s" % {'err_type': error.exc_type,
'err_msg': error.value}
raise webob.exc.HTTPBadRequest(explanation=msg)
except Exception as error:
- raise webob.exc.HTTPBadRequest(explanation=unicode(error))
+ raise webob.exc.HTTPBadRequest(explanation=six.text_type(error))
return {'os-volume_upload_image': response}
@wsgi.action('os-extend')
# under the License.
from oslo_log import log as logging
+import six
import webob
from webob import exc
msg = _("Volume could not be found")
raise exc.HTTPNotFound(explanation=msg)
except exception.ReplicationError as error:
- raise exc.HTTPBadRequest(explanation=unicode(error))
+ raise exc.HTTPBadRequest(explanation=six.text_type(error))
return webob.Response(status_int=202)
@wsgi.response(202)
msg = _("Volume could not be found")
raise exc.HTTPNotFound(explanation=msg)
except exception.ReplicationError as error:
- raise exc.HTTPBadRequest(explanation=unicode(error))
+ raise exc.HTTPBadRequest(explanation=six.text_type(error))
return webob.Response(status_int=202)
# under the License.
from oslo_log import log as logging
+import six
import webob.dec
import webob.exc
# including those that are safe to expose, see bug 1021373
if safe:
msg = (inner.msg if isinstance(inner, exception.CinderException)
- else unicode(inner))
+ else six.text_type(inner))
params = {'exception': inner.__class__.__name__,
'explanation': msg}
outer.explanation = _('%(exception)s: %(explanation)s') % params
import re
from lxml import etree
+import six
from cinder.i18n import _
from cinder import utils
# Start with the text...
if self.text is not None:
- elem.text = unicode(self.text(obj))
+ elem.text = six.text_type(self.text(obj))
# Now set up all the attributes...
for key, value in self.attrib.items():
try:
- elem.set(key, unicode(value(obj, True)))
+ elem.set(key, six.text_type(value(obj, True)))
except KeyError:
# Attribute has no value, so don't include it
pass
pass
err = _('unsupported compression algorithm: %s') % algorithm
- raise ValueError(unicode(err))
+ raise ValueError(err)
def __init__(self, context, chunk_size_bytes, sha_block_size_bytes,
backup_default_container, enable_progress_timer,
p1 = subprocess.Popen(cmd1, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
except OSError as e:
- LOG.error(_LE("Pipe1 failed - %s ") % unicode(e))
+ LOG.error(_LE("Pipe1 failed - %s "), e)
raise
# NOTE(dosaboy): ensure that the pipe is blocking. This is to work
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
except OSError as e:
- LOG.error(_LE("Pipe2 failed - %s ") % unicode(e))
+ LOG.error(_LE("Pipe2 failed - %s "), e)
raise
p1.stdout.close()
import oslo_messaging as messaging
from oslo_utils import excutils
from oslo_utils import importutils
+import six
from cinder.backup import driver
from cinder.backup import rpcapi as backup_rpcapi
{'status': 'available'})
self.db.backup_update(context, backup_id,
{'status': 'error',
- 'fail_reason': unicode(err)})
+ 'fail_reason': six.text_type(err)})
self.db.volume_update(context, volume_id, {'status': 'available'})
backup = self.db.backup_update(context, backup_id,
self.db.backup_update(context, backup_id,
{'status': 'error',
'fail_reason':
- unicode(err)})
+ six.text_type(err)})
LOG.info(_LI('Delete backup started, backup: %s.'), backup_id)
backup = self.db.backup_get(context, backup_id)
self.db.backup_update(context, backup_id,
{'status': 'error',
'fail_reason':
- unicode(err)})
+ six.text_type(err)})
# Get reservations
try:
backup_url = backup_service.export_record(backup)
backup_record['backup_url'] = backup_url
except Exception as err:
- msg = unicode(err)
+ msg = six.text_type(err)
raise exception.InvalidBackup(reason=msg)
LOG.info(_LI('Export record finished, backup %s exported.'), backup_id)
backup_service = self.service.get_backup_driver(context)
backup_options = backup_service.import_record(backup_url)
except Exception as err:
- msg = unicode(err)
+ msg = six.text_type(err)
self.db.backup_update(context,
backup_id,
{'status': 'error',
self.db.backup_update(context, backup_id,
{'status': 'error',
'fail_reason':
- unicode(err)})
+ six.text_type(err)})
LOG.info(_LI('Import record id %s metadata from driver '
'finished.') % backup_id)
"""Exceptions for the Brick library."""
from oslo_log import log as logging
+import six
from cinder.i18n import _
super(BrickException, self).__init__(message)
def __unicode__(self):
- return unicode(self.msg)
+ return six.text_type(self.msg)
class NotFound(BrickException):
return self.kwargs['message'] is None or '%(message)' in self.message
def __unicode__(self):
- return unicode(self.msg)
+ return six.text_type(self.msg)
class VolumeBackendAPIException(CinderException):
yield(0, msg)
+def check_unicode_usage(logical_line, noqa):
+ if noqa:
+ return
+
+ msg = "C302: Found unicode() call. Please use six.text_type()."
+
+ if 'unicode(' in logical_line:
+ yield(0, msg)
+
+
def factory(register):
register(no_vi_headers)
register(no_translate_debug_logs)
register(check_no_contextlib_nested)
register(check_datetime_now)
register(validate_log_translations)
+ register(check_unicode_usage)
# FIXME(danms): We should really try to avoid the need to do this
if isinstance(value, (six.string_types, int, long, float,
datetime.datetime)):
- return unicode(value)
+ return six.text_type(value)
else:
raise ValueError(_('A string is required here, not %s') %
value.__class__.__name__)
from oslo_concurrency import processutils as putils
from oslo_config import cfg
from oslo_log import log as logging
+import six
from cinder.brick import exception
from cinder.brick.initiator import connector
name = 'volume-00000001'
vol = {'id': 1, 'name': name}
# Should work for string, unicode, and list
- wwns = ['1234567890123456', unicode('1234567890123456'),
+ wwns = ['1234567890123456', six.text_type('1234567890123456'),
['1234567890123456', '1234567890123457']]
for wwn in wwns:
connection_info = self.fibrechan_connection(vol, location, wwn)
from cinder.brick import exception
from cinder import test
+import six
+
class BrickExceptionTestCase(test.TestCase):
def test_default_error_msg(self):
message = "default message"
exc = FakeBrickException()
- self.assertEqual(unicode(exc), 'default message')
+ self.assertEqual(six.text_type(exc), 'default message')
def test_error_msg(self):
- self.assertEqual(unicode(exception.BrickException('test')), 'test')
+ self.assertEqual(six.text_type(exception.BrickException('test')),
+ 'test')
def test_default_error_msg_with_kwargs(self):
class FakeBrickException(exception.BrickException):
message = "default message: %(code)s"
exc = FakeBrickException(code=500)
- self.assertEqual(unicode(exc), 'default message: 500')
+ self.assertEqual(six.text_type(exc), 'default message: 500')
def test_error_msg_exception_with_kwargs(self):
# NOTE(dprince): disable format errors for this test
message = "default message: %(mispelled_code)s"
exc = FakeBrickException(code=500)
- self.assertEqual(unicode(exc), 'default message: %(mispelled_code)s')
+ self.assertEqual(six.text_type(exc),
+ 'default message: %(mispelled_code)s')
def test_default_error_code(self):
class FakeBrickException(exception.BrickException):
from cinder import exception
from cinder import test
+import six
+
class FakeNotifier(object):
"""Acts like the cinder.openstack.common.notifier.api module."""
message = "default message"
exc = FakeCinderException()
- self.assertEqual(unicode(exc), 'default message')
+ self.assertEqual(six.text_type(exc), 'default message')
def test_error_msg(self):
- self.assertEqual(unicode(exception.CinderException('test')), 'test')
+ self.assertEqual(six.text_type(exception.CinderException('test')),
+ 'test')
def test_default_error_msg_with_kwargs(self):
class FakeCinderException(exception.CinderException):
message = "default message: %(code)s"
exc = FakeCinderException(code=500)
- self.assertEqual(unicode(exc), 'default message: 500')
+ self.assertEqual(six.text_type(exc), 'default message: 500')
def test_error_msg_exception_with_kwargs(self):
# NOTE(dprince): disable format errors for this test
message = "default message: %(misspelled_code)s"
exc = FakeCinderException(code=500)
- self.assertEqual(unicode(exc), 'default message: %(misspelled_code)s')
+ self.assertEqual(six.text_type(exc),
+ 'default message: %(misspelled_code)s')
def test_default_error_code(self):
class FakeCinderException(exception.CinderException):
message = 'FakeCinderException: %(message)s'
exc = FakeCinderException(message='message')
- self.assertEqual(unicode(exc), 'FakeCinderException: message')
+ self.assertEqual(six.text_type(exc), 'FakeCinderException: message')
def test_message_and_kwarg_in_format_string(self):
class FakeCinderException(exception.CinderException):
message = 'Error %(code)d: %(message)s'
exc = FakeCinderException(message='message', code=404)
- self.assertEqual(unicode(exc), 'Error 404: message')
+ self.assertEqual(six.text_type(exc), 'Error 404: message')
def test_message_is_exception_in_format_string(self):
class FakeCinderException(exception.CinderException):
msg = 'test message'
exc1 = Exception(msg)
exc2 = FakeCinderException(message=exc1)
- self.assertEqual(unicode(exc2), 'Exception: test message')
+ self.assertEqual(six.text_type(exc2), 'Exception: test message')
"LOG.error(_LE('foo')", "foo.py"))))
self.assertEqual(0, len(list(checks.validate_log_translations(
"LOG.exception(_LE('foo')", "foo.py"))))
+
+ def test_check_unicode_usage(self):
+ self.assertEqual(1, len(list(checks.check_unicode_usage(
+ "unicode(msg)", False))))
+ self.assertEqual(0, len(list(checks.check_unicode_usage(
+ "unicode(msg) # noqa", True))))
import mock
from oslo_log import log as logging
from oslo_utils import units
+import six
from cinder import context
from cinder import exception
def test_paramiko_1_13_0(cliq_args):
# paramiko 1.13.0 now returns unicode
- output = unicode(
+ output = six.text_type(
'<?xml version="1.0" encoding="UTF-8" standalone="no" ?>\n'
'<gauche version="1.0">\n\n <response description="Operation'
' succeeded." name="CliqSuccess" processingTime="423" '
invalid_ch_in_host = invalid_ch_in_host + ch
host_name = connector['host']
- if isinstance(host_name, unicode):
- unicode_host_name_filter = dict((ord(unicode(char)), u'-')
+ if isinstance(host_name, six.text_type):
+ unicode_host_name_filter = dict((ord(six.text_type(char)), u'-')
for char in invalid_ch_in_host)
host_name = host_name.translate(unicode_host_name_filter)
elif isinstance(host_name, str):
from oslo_utils import timeutils
from oslo_utils import uuidutils
from osprofiler import profiler
+import six
from taskflow import exceptions as tfe
from cinder import compute
self._delete_image(context, image_meta['id'], image_service)
with excutils.save_and_reraise_exception():
- payload['message'] = unicode(error)
+ payload['message'] = six.text_type(error)
finally:
if not volume['volume_attachment']:
self.db.volume_update(context, volume_id,
Code Review / openstack-build / cinder-build.git / commitdiff