- [C305] Prevent use of deprecated contextlib.nested.
- [C306] timeutils.strtime() must not be used (deprecated).
- [C307] LOG.warn is deprecated. Enforce use of LOG.warning.
+- [C308] timeutils.isotime() must not be used (deprecated).
General
-------
import datetime
-from oslo_utils import timeutils
-
class ViewBuilder(object):
"""OpenStack API base limits view builder."""
"value": rate_limit["value"],
"remaining": int(rate_limit["remaining"]),
"unit": rate_limit["unit"],
- "next-available": timeutils.isotime(at=next_avail),
+ "next-available": next_avail.isoformat(),
}
'read_deleted': self.read_deleted,
'roles': self.roles,
'remote_address': self.remote_address,
- 'timestamp': timeutils.isotime(self.timestamp, True),
+ 'timestamp': self.timestamp.isoformat(),
'quota_class': self.quota_class,
'service_catalog': self.service_catalog,
'request_id': self.request_id}
def check_timeutils_strtime(logical_line):
msg = ("C306: Found timeutils.strtime(). "
- "Please use oslo_utils.timeutils.isotime() or datetime.strftime()")
+ "Please use datetime.datetime.isoformat() or datetime.strftime()")
if 'timeutils.strtime' in logical_line:
yield(0, msg)
yield (0, msg)
+def check_timeutils_isotime(logical_line):
+ msg = ("C308: Found timeutils.isotime(). "
+ "Please use datetime.datetime.isoformat()")
+ if 'timeutils.isotime' in logical_line:
+ yield(0, msg)
+
+
def factory(register):
register(no_vi_headers)
register(no_translate_debug_logs)
register(check_oslo_namespace_imports)
register(check_datetime_now)
register(check_timeutils_strtime)
+ register(check_timeutils_isotime)
register(validate_log_translations)
register(check_unicode_usage)
register(check_no_print_statements)
import traceback
from oslo_log import log as logging
-from oslo_utils import timeutils
from oslo_versionedobjects import base
from oslo_versionedobjects import fields
import six
not isinstance(value_arg, six.string_types) and value_arg):
kwargs[kw] = ''.join(traceback.format_tb(value_arg))
elif isinstance(value_arg, datetime.datetime):
- kwargs[kw] = timeutils.isotime(value_arg)
+ kwargs[kw] = value_arg.isoformat()
if hasattr(fn, '__call__'):
return fn(obj, *args, **kwargs)
# NOTE(danms): We wrap a descriptor, so use that protocol
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 10,
"remaining": 10,
},
{
"verb": "POST",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "HOUR",
"value": 5,
"remaining": 5,
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 5,
"remaining": 5,
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 10,
"remaining": 10,
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 10,
"remaining": 10,
"injected_file_content_bytes": 5}
def test_build_limits(self):
- tdate = "2011-07-21T18:17:06Z"
+ tdate = "2011-07-21T18:17:06"
expected_limits = \
{"limits": {"rate": [{"uri": "*",
"regex": ".*",
def test_view_builder_show(self):
view_builder = views_types.ViewBuilder()
- now = timeutils.isotime()
+ now = timeutils.utcnow().isoformat()
raw_volume_type = dict(name='new_type',
deleted=False,
created_at=now,
def test_view_builder_list(self):
view_builder = views_types.ViewBuilder()
- now = timeutils.isotime()
+ now = timeutils.utcnow().isoformat()
raw_volume_types = []
for i in range(0, 10):
raw_volume_types.append(dict(name='new_type',
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 10,
"remaining": 10,
},
{
"verb": "POST",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "HOUR",
"value": 5,
"remaining": 5,
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 5,
"remaining": 5,
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 10,
"remaining": 10,
"limit": [
{
"verb": "GET",
- "next-available": "1970-01-01T00:00:00Z",
+ "next-available": "1970-01-01T00:00:00",
"unit": "MINUTE",
"value": 10,
"remaining": 10,
"injected_file_content_bytes": 5}
def test_build_limits(self):
- tdate = "2011-07-21T18:17:06Z"
+ tdate = "2011-07-21T18:17:06"
expected_limits = {
"limits": {"rate": [{"uri": "*",
"regex": ".*",
def test_view_builder_show(self):
view_builder = views_types.ViewBuilder()
- now = timeutils.isotime()
+ now = timeutils.utcnow().isoformat()
raw_volume_type = dict(
name='new_type',
description='new_type_desc',
def test_view_builder_list(self):
view_builder = views_types.ViewBuilder()
- now = timeutils.isotime()
+ now = timeutils.utcnow().isoformat()
raw_volume_types = []
for i in range(0, 10):
raw_volume_types.append(
# to set any that were provided
params = {'volumeID': sf_volume_id}
- create_time = timeutils.isotime(v_ref['created_at'], True)
+ create_time = v_ref['created_at'].isoformat()
attributes = {'uuid': v_ref['id'],
'is_clone': 'True',
'src_uuid': src_uuid,
if type_id is not None:
qos = self._set_qos_by_volume_type(ctxt, type_id)
- create_time = timeutils.isotime(volume['created_at'], True)
+ create_time = volume['created_at'].isoformat()
attributes = {'uuid': volume['id'],
'is_clone': 'False',
'created_at': create_time}
raise exception.VolumeNotFound(volume_id=volume['id'])
attributes = sf_vol['attributes']
- attributes['retyped_at'] = timeutils.isotime(subsecond=True)
+ attributes['retyped_at'] = timeutils.utcnow().isoformat()
params = {'volumeID': sf_vol['volumeID']}
qos = self._set_qos_by_volume_type(ctxt, new_type['id'])
if type_id is not None:
qos = self._set_qos_by_volume_type(ctxt, type_id)
- import_time = timeutils.isotime(volume['created_at'], True)
+ import_time = volume['created_at'].isoformat()
attributes = {'uuid': volume['id'],
'is_clone': 'False',
'os_imported_at': import_time,
if sf_vol is None:
raise exception.VolumeNotFound(volume_id=volume['id'])
- export_time = timeutils.isotime(subsecond=True)
+ export_time = timeutils.utcnow().isoformat()
attributes = sf_vol['attributes']
attributes['os_exported_at'] = export_time
params = {'volumeID': int(sf_vol['volumeID']),
def _usage_from_volume(volume_ref, **kw):
+ now = timeutils.utcnow()
+ launched_at = volume_ref['launched_at'] or now
+ created_at = volume_ref['created_at'] or now
usage_info = dict(
tenant_id=volume_ref['project_id'],
host=volume_ref['host'],
volume_id=volume_ref['id'],
volume_type=volume_ref['volume_type_id'],
display_name=volume_ref['display_name'],
- launched_at=timeutils.isotime(at=volume_ref['launched_at']),
- created_at=timeutils.isotime(at=volume_ref['created_at']),
+ launched_at=launched_at.isoformat(),
+ created_at=created_at.isoformat(),
status=volume_ref['status'],
snapshot_id=volume_ref['snapshot_id'],
size=volume_ref['size'],
availability_zone=group_ref['availability_zone'],
consistencygroup_id=group_ref['id'],
name=group_ref['name'],
- created_at=timeutils.isotime(at=group_ref['created_at']),
+ created_at=group_ref['created_at'].isoformat(),
status=group_ref['status'])
usage_info.update(kw)
cgsnapshot_id=cgsnapshot_ref['id'],
name=cgsnapshot_ref['name'],
consistencygroup_id=cgsnapshot_ref['consistencygroup_id'],
- created_at=timeutils.isotime(at=cgsnapshot_ref['created_at']),
+ created_at=cgsnapshot_ref['created_at'].isoformat(),
status=cgsnapshot_ref['status'])
usage_info.update(kw)