- [N327] assert_called_once is not a valid Mock method.
- [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().
General
LOG.info(_LI('Purging deleted rows older than age=%(age)d days '
'from table=%(table)s'), {'age': age_in_days,
'table': table})
- deleted_age = dt.datetime.now() - dt.timedelta(days=age_in_days)
+ deleted_age = timeutils.utcnow() - dt.timedelta(days=age_in_days)
try:
with session.begin():
result = session.execute(
LOG = logging.getLogger(__name__)
CLASS_NAME = 'default'
-CREATED_AT = datetime.datetime.now()
+CREATED_AT = datetime.datetime.now() # noqa
def upgrade(migrate_engine):
LOG = logging.getLogger(__name__)
CLASS_NAME = 'default'
-CREATED_AT = datetime.datetime.now()
+CREATED_AT = datetime.datetime.now() # noqa
def upgrade(migrate_engine):
LOG = logging.getLogger(__name__)
-CREATED_AT = datetime.datetime.now()
+CREATED_AT = datetime.datetime.now() # noqa
def upgrade(migrate_engine):
yield(0, msg)
+def check_datetime_now(logical_line, noqa):
+ if noqa:
+ return
+
+ msg = ("C301: Found datetime.now(). "
+ "Please use timeutils.utcnow() from oslo_utils.")
+ if 'datetime.now' in logical_line:
+ yield(0, msg)
+
+
def factory(register):
register(no_vi_headers)
register(no_translate_debug_logs)
register(check_assert_called_once)
register(check_oslo_namespace_imports)
register(check_no_contextlib_nested)
+ register(check_datetime_now)
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
import json
import uuid
from lxml import etree
+from oslo_utils import timeutils
import webob
from cinder import context
'status': 'available',
'size': 5,
'availability_zone': 'somewhere',
- 'created_at': datetime.datetime.now(),
+ 'created_at': timeutils.utcnow(),
'attach_status': None,
'display_name': 'anothervolume',
'display_description': 'Just another volume!',
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
import json
import uuid
from xml.dom import minidom
+from oslo_utils import timeutils
import webob
from cinder.api import common
'status': 'available',
'size': 5,
'availability_zone': 'somewhere',
- 'created_at': datetime.datetime.now(),
+ 'created_at': timeutils.utcnow(),
'attach_status': None,
'display_name': 'anothervolume',
'display_description': 'Just another volume!',
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
import json
import uuid
from lxml import etree
+from oslo_utils import timeutils
import webob
from cinder import context
'status': 'available',
'size': 5,
'availability_zone': 'somewhere',
- 'created_at': datetime.datetime.now(),
+ 'created_at': timeutils.utcnow(),
'attach_status': None,
'display_name': 'anothervolume',
'display_description': 'Just another volume!',
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
import json
import uuid
from lxml import etree
+from oslo_utils import timeutils
import webob
from cinder import context
'status': 'available',
'size': 5,
'availability_zone': 'somewhere',
- 'created_at': datetime.datetime.now(),
+ 'created_at': timeutils.utcnow(),
'attach_status': None,
'display_name': 'anothervolume',
'display_description': 'Just another volume!',
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
-
from lxml import etree
import mock
from oslo_log import log as logging
+from oslo_utils import timeutils
import webob
from cinder.api.v1 import snapshots
id='snap_id',
status='snap_status',
size=1024,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
display_name='snap_name',
display_description='snap_desc',
volume_id='vol_id', )
raw_snapshots = [dict(id='snap1_id',
status='snap1_status',
size=1024,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
display_name='snap1_name',
display_description='snap1_desc',
volume_id='vol1_id', ),
dict(id='snap2_id',
status='snap2_status',
size=1024,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
display_name='snap2_name',
display_description='snap2_desc',
volume_id='vol2_id', )]
from lxml import etree
import mock
from oslo_config import cfg
+from oslo_utils import timeutils
import webob
from cinder.api import extensions
size=1024,
availability_zone='vol_availability',
bootable='false',
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
attachments=[dict(id='vol_id',
volume_id='vol_id',
server_id='instance_uuid',
size=1024,
availability_zone='vol1_availability',
bootable='true',
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
attachments=[dict(id='vol1_id',
volume_id='vol1_id',
server_id='instance_uuid',
size=1024,
availability_zone='vol2_availability',
bootable='true',
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
attachments=[dict(id='vol2_id',
volume_id='vol2_id',
server_id='instance_uuid',
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
-
from lxml import etree
import mock
from oslo_log import log as logging
+from oslo_utils import timeutils
import webob
from cinder.api.v2 import snapshots
id='snap_id',
status='snap_status',
size=1024,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
name='snap_name',
description='snap_desc',
display_description='snap_desc',
id='snap1_id',
status='snap1_status',
size=1024,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
name='snap1_name',
description='snap1_desc',
volume_id='vol1_id',
id='snap2_id',
status='snap2_status',
size=1024,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
name='snap2_name',
description='snap2_desc',
volume_id='vol2_id',
from lxml import etree
import mock
from oslo_config import cfg
+from oslo_utils import timeutils
import six
import six.moves.urllib.parse as urlparse
import webob
size=1024,
availability_zone='vol_availability',
bootable=False,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
attachments=[
dict(
id='vol_id',
size=1024,
availability_zone='vol1_availability',
bootable=True,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
attachments=[
dict(
id='vol1_id',
size=1024,
availability_zone='vol2_availability',
bootable=False,
- created_at=datetime.datetime.now(),
+ created_at=timeutils.utcnow(),
attachments=[dict(id='vol2_id',
volume_id='vol2_id',
server_id='instance_uuid',
import uuid
from oslo_log import log as logging
+from oslo_utils import timeutils
from cinder import context
from cinder import db
ins_stmt = self.vm.insert().values(volume_id=uuidstr)
self.conn.execute(ins_stmt)
# Set 4 of them deleted, 2 are 60 days ago, 2 are 20 days ago
- old = datetime.datetime.now() - datetime.timedelta(days=20)
- older = datetime.datetime.now() - datetime.timedelta(days=60)
+ old = timeutils.utcnow() - datetime.timedelta(days=20)
+ older = timeutils.utcnow() - datetime.timedelta(days=60)
make_old = self.volumes.update().\
where(self.volumes.c.id.in_(self.uuidstrs[1:3]))\
.values(deleted_at=old)
Tests for volume backup to IBM Tivoli Storage Manager (TSM).
"""
-import datetime
import json
import os
import posix
from oslo_concurrency import processutils as putils
from oslo_log import log as logging
+from oslo_utils import timeutils
from cinder.backup.drivers import tsm
from cinder import context
self._backup_list[path] = []
else:
self._backup_list[path][-1]['active'] = False
- date = datetime.datetime.now()
+ date = timeutils.utcnow()
datestr = date.strftime("%m/%d/%Y %H:%M:%S")
self._backup_list[path].append({'date': datestr, 'active': True})
retcode = 0
"with contextlib.nested("))))
self.assertEqual(0, len(list(checks.check_no_contextlib_nested(
"with foo as bar"))))
+
+ def test_check_datetime_now(self):
+ self.assertEqual(1, len(list(checks.check_datetime_now(
+ "datetime.now", False))))
+ self.assertEqual(0, len(list(checks.check_datetime_now(
+ "timeutils.utcnow", False))))
+
+ def test_check_datetime_now_noqa(self):
+ self.assertEqual(0, len(list(checks.check_datetime_now(
+ "datetime.now() # noqa", True))))
\ No newline at end of file
def _create_snapshot_name(self):
"""Creates a snapshot name from the date and time."""
- return 'cinder-zfssa-nfs-snapshot-%s' % dt.datetime.now().isoformat()
+
+ return ('cinder-zfssa-nfs-snapshot-%s' %
+ dt.datetime.utcnow().isoformat())
def _get_share_capacity_info(self):
"""Get available and used capacity info for the NFS share."""