Replace deprecated datetime.utcnow()

The datetime.utcnow() is deprecated in Python 3.12.
Replace datetime.utcnow() with oslo_utils.timeutils.utcnow().
This bumps oslo.utils to 7.0.0.

Change-Id: Ic4e33df4fc9e858a9de139066feaaaec71bfc6fe
Signed-off-by: Takashi Natsume <takanattie@gmail.com>
This commit is contained in:
Takashi Natsume 2024-10-06 00:22:09 +09:00
parent 2ec8724585
commit b365afcd9c
13 changed files with 44 additions and 43 deletions

View File

@ -16,10 +16,10 @@
""" """
Retry/scheduler classes and logic. Retry/scheduler classes and logic.
""" """
import datetime
import random import random
from oslo_service import service from oslo_service import service
from oslo_utils import timeutils
from barbican.common import config from barbican.common import config
from barbican.common import utils from barbican.common import utils
@ -118,7 +118,7 @@ class PeriodicServer(service.Service):
repositories.start() repositories.start()
try: try:
entities, _, _, total = self.order_retry_repo.get_by_create_date( entities, _, _, total = self.order_retry_repo.get_by_create_date(
only_at_or_before_this_date=datetime.datetime.utcnow(), only_at_or_before_this_date=timeutils.utcnow(),
suppress_exception=True) suppress_exception=True)
finally: finally:
repositories.clear() repositories.clear()

View File

@ -27,6 +27,7 @@ except ImportError:
newrelic_loaded = False newrelic_loaded = False
from oslo_service import service from oslo_service import service
from oslo_utils import timeutils
from barbican.common import utils from barbican.common import utils
from barbican.model import models from barbican.model import models
@ -166,7 +167,7 @@ def schedule_order_retry_tasks(
LOG.debug( LOG.debug(
'Scheduling RPC method for retry: {0}'.format(retry_rpc_method)) 'Scheduling RPC method for retry: {0}'.format(retry_rpc_method))
date_to_retry_at = datetime.datetime.utcnow() + datetime.timedelta( date_to_retry_at = timeutils.utcnow() + datetime.timedelta(
milliseconds=retry_result.retry_msec) milliseconds=retry_result.retry_msec)
retry_model = models.OrderRetryTask() retry_model = models.OrderRetryTask()

View File

@ -17,6 +17,7 @@ import datetime
from unittest import mock from unittest import mock
from oslo_db import exception as db_exc from oslo_db import exception as db_exc
from oslo_utils import timeutils
from sqlalchemy.exc import IntegrityError from sqlalchemy.exc import IntegrityError
from barbican.model import clean from barbican.model import clean
@ -237,7 +238,7 @@ class WhenTestingDBCleanUpCommand(utils.RepositoryTestCase):
"""Test cleaning up entries within date""" """Test cleaning up entries within date"""
secret = _setup_entry('secret', project=project) secret = _setup_entry('secret', project=project)
order = order = _setup_entry('order', project=project, secret=secret) order = order = _setup_entry('order', project=project, secret=secret)
current_time = datetime.datetime.utcnow() current_time = timeutils.utcnow()
tomorrow = current_time + datetime.timedelta(days=1) tomorrow = current_time + datetime.timedelta(days=1)
yesterday = current_time - datetime.timedelta(days=1) yesterday = current_time - datetime.timedelta(days=1)
secret.delete() secret.delete()
@ -261,7 +262,7 @@ class WhenTestingDBCleanUpCommand(utils.RepositoryTestCase):
def test_soft_deleting_expired_secrets(self, project): def test_soft_deleting_expired_secrets(self, project):
"""Test soft deleting secrets that are expired""" """Test soft deleting secrets that are expired"""
current_time = datetime.datetime.utcnow() current_time = timeutils.utcnow()
tomorrow = current_time + datetime.timedelta(days=1) tomorrow = current_time + datetime.timedelta(days=1)
yesterday = current_time - datetime.timedelta(days=1) yesterday = current_time - datetime.timedelta(days=1)

View File

@ -17,6 +17,7 @@ import datetime
import unittest import unittest
from oslo_serialization import base64 from oslo_serialization import base64
from oslo_utils import timeutils
from barbican.common import exception as excep from barbican.common import exception as excep
from barbican.common import validators from barbican.common import validators
@ -1203,7 +1204,7 @@ class WhenTestingKeyTypeOrderValidator(utils.BaseTestCase):
def test_should_pass_good_exp_meta_in_order_refs(self): def test_should_pass_good_exp_meta_in_order_refs(self):
self.key_order_req['meta']['algorithm'] = 'AES' self.key_order_req['meta']['algorithm'] = 'AES'
ony_year_factor = datetime.timedelta(days=1 * 365) ony_year_factor = datetime.timedelta(days=1 * 365)
date_after_year = datetime.datetime.utcnow() + ony_year_factor date_after_year = timeutils.utcnow() + ony_year_factor
date_after_year_str = date_after_year.strftime('%Y-%m-%d %H:%M:%S') date_after_year_str = date_after_year.strftime('%Y-%m-%d %H:%M:%S')
self.key_order_req['meta']['expiration'] = date_after_year_str self.key_order_req['meta']['expiration'] = date_after_year_str
result = self.validator.validate(self.key_order_req) result = self.validator.validate(self.key_order_req)

View File

@ -17,9 +17,8 @@ Supports database/repositories oriented unit testing.
Warning: Do not merge this content with the utils.py module, as doing so will Warning: Do not merge this content with the utils.py module, as doing so will
break the DevStack functional test discovery process. break the DevStack functional test discovery process.
""" """
import datetime from oslo_utils import timeutils
import oslotest.base as oslotest import oslotest.base as oslotest
from sqlalchemy.engine import Engine from sqlalchemy.engine import Engine
from sqlalchemy import event from sqlalchemy import event
@ -168,7 +167,7 @@ def create_order_retry(order=None, retry_task="", retry_args=[],
order_retry.retry_args = retry_args order_retry.retry_args = retry_args
order_retry.retry_kwargs = retry_kwargs order_retry.retry_kwargs = retry_kwargs
if not retry_at: if not retry_at:
order_retry.retry_at = datetime.datetime.utcnow() order_retry.retry_at = timeutils.utcnow()
order_retry.order_id = order.id order_retry.order_id = order.id
order_retry_task_repo = repositories.get_order_retry_tasks_repository() order_retry_task_repo = repositories.get_order_retry_tasks_repository()
order_retry_task_repo.create_from(order_retry, session) order_retry_task_repo.create_from(order_retry, session)

View File

@ -12,6 +12,8 @@
import datetime import datetime
from oslo_utils import timeutils
from barbican.common import exception from barbican.common import exception
from barbican.common import resources as res from barbican.common import resources as res
from barbican.model import models from barbican.model import models
@ -24,8 +26,7 @@ class WhenTestingCertificateAuthorityRepo(database_utils.RepositoryTestCase):
def setUp(self): def setUp(self):
super(WhenTestingCertificateAuthorityRepo, self).setUp() super(WhenTestingCertificateAuthorityRepo, self).setUp()
self.ca_repo = repositories.CertificateAuthorityRepo() self.ca_repo = repositories.CertificateAuthorityRepo()
expiration = (datetime.datetime.utcnow() + expiration = timeutils.utcnow() + datetime.timedelta(minutes=10)
datetime.timedelta(minutes=10))
self.parsed_ca = {'plugin_name': 'dogtag_plugin', self.parsed_ca = {'plugin_name': 'dogtag_plugin',
'plugin_ca_id': 'ca_master', 'plugin_ca_id': 'ca_master',
'name': 'Dogtag CA', 'name': 'Dogtag CA',
@ -182,8 +183,7 @@ class WhenTestingProjectCARepo(database_utils.RepositoryTestCase):
super(WhenTestingProjectCARepo, self).setUp() super(WhenTestingProjectCARepo, self).setUp()
self.ca_repo = repositories.CertificateAuthorityRepo() self.ca_repo = repositories.CertificateAuthorityRepo()
self.project_ca_repo = repositories.ProjectCertificateAuthorityRepo() self.project_ca_repo = repositories.ProjectCertificateAuthorityRepo()
expiration = (datetime.datetime.utcnow() + expiration = timeutils.utcnow() + datetime.timedelta(minutes=10)
datetime.timedelta(minutes=10))
self.parsed_ca = {'plugin_name': 'dogtag_plugin', self.parsed_ca = {'plugin_name': 'dogtag_plugin',
'plugin_ca_id': 'ca_master', 'plugin_ca_id': 'ca_master',
'expiration': expiration.isoformat(), 'expiration': expiration.isoformat(),
@ -305,11 +305,9 @@ class WhenTestingPreferredCARepo(database_utils.RepositoryTestCase):
self.preferred_ca_repo = ( self.preferred_ca_repo = (
repositories.PreferredCertificateAuthorityRepo()) repositories.PreferredCertificateAuthorityRepo())
expiration = (datetime.datetime.utcnow() + expiration = timeutils.utcnow() + datetime.timedelta(minutes=10)
datetime.timedelta(minutes=10))
expiration_later = (datetime.datetime.utcnow() + expiration_later = timeutils.utcnow() + datetime.timedelta(days=10)
datetime.timedelta(days=10))
self.parsed_ca = {'plugin_name': 'dogtag_plugin', self.parsed_ca = {'plugin_name': 'dogtag_plugin',
'plugin_ca_id': 'ca_master', 'plugin_ca_id': 'ca_master',
'expiration': expiration.isoformat(), 'expiration': expiration.isoformat(),

View File

@ -15,6 +15,8 @@
import datetime import datetime
import time import time
from oslo_utils import timeutils
from barbican.common import config from barbican.common import config
from barbican.common import exception from barbican.common import exception
from barbican.model import models from barbican.model import models
@ -27,7 +29,7 @@ class WhenTestingOrderRetryTaskRepository(database_utils.RepositoryTestCase):
def setUp(self): def setUp(self):
super(WhenTestingOrderRetryTaskRepository, self).setUp() super(WhenTestingOrderRetryTaskRepository, self).setUp()
self.date_time_now = datetime.datetime.utcnow() self.date_time_now = timeutils.utcnow()
self.test_args = ['test', 'args'] self.test_args = ['test', 'args']
self.test_kwargs = {'test': 1, 'kwargs': 2} self.test_kwargs = {'test': 1, 'kwargs': 2}
@ -78,7 +80,7 @@ class WhenTestingOrderRetryTaskRepository(database_utils.RepositoryTestCase):
# Now, a retrieve by the current time should return our entry. # Now, a retrieve by the current time should return our entry.
entities, offset, limit, total = self.repo.get_by_create_date( entities, offset, limit, total = self.repo.get_by_create_date(
only_at_or_before_this_date=datetime.datetime.utcnow(), only_at_or_before_this_date=timeutils.utcnow(),
session=session, session=session,
suppress_exception=True suppress_exception=True
) )

View File

@ -13,6 +13,7 @@
import datetime import datetime
import fixtures import fixtures
from oslo_utils import timeutils
import testtools import testtools
from barbican.common import exception from barbican.common import exception
@ -222,7 +223,7 @@ class WhenTestingSecretRepository(database_utils.RepositoryTestCase):
self.assertEqual(1, count) self.assertEqual(1, count)
def test_should_get_count_one_after_expiration(self): def test_should_get_count_one_after_expiration(self):
current_time = datetime.datetime.utcnow() current_time = timeutils.utcnow()
tomorrow = current_time + datetime.timedelta(days=1) tomorrow = current_time + datetime.timedelta(days=1)
yesterday = current_time - datetime.timedelta(days=1) yesterday = current_time - datetime.timedelta(days=1)

View File

@ -16,6 +16,8 @@
import datetime import datetime
import unittest import unittest
from oslo_utils import timeutils
from barbican.common import exception from barbican.common import exception
from barbican.model import models from barbican.model import models
from barbican.plugin.interface import secret_store from barbican.plugin.interface import secret_store
@ -36,7 +38,7 @@ class WhenCreatingNewSecret(utils.BaseTestCase):
self.parsed_order = {'secret': self.parsed_secret} self.parsed_order = {'secret': self.parsed_secret}
def test_new_secret_is_created_from_dict(self): def test_new_secret_is_created_from_dict(self):
date_time = datetime.datetime.utcnow().isoformat() date_time = timeutils.utcnow().isoformat()
self.parsed_secret['expiration'] = date_time self.parsed_secret['expiration'] = date_time
secret = models.Secret(self.parsed_secret) secret = models.Secret(self.parsed_secret)
self.assertEqual(self.parsed_secret['name'], secret.name) self.assertEqual(self.parsed_secret['name'], secret.name)
@ -57,7 +59,7 @@ class WhenCreatingNewSecret(utils.BaseTestCase):
def test_new_secret_is_created_with_default_secret_type(self): def test_new_secret_is_created_with_default_secret_type(self):
secret_spec = dict(self.parsed_secret) secret_spec = dict(self.parsed_secret)
date_time = datetime.datetime.utcnow().isoformat() date_time = timeutils.utcnow().isoformat()
secret_spec['expiration'] = date_time secret_spec['expiration'] = date_time
del secret_spec['secret_type'] del secret_spec['secret_type']
secret = models.Secret(secret_spec) secret = models.Secret(secret_spec)
@ -267,7 +269,7 @@ class WhenCreatingOrderRetryTask(utils.BaseTestCase):
'sub_status': 'Pending', 'sub_status': 'Pending',
'sub_status_message': 'Waiting for instructions...' 'sub_status_message': 'Waiting for instructions...'
}) })
at = datetime.datetime.utcnow() at = timeutils.utcnow()
order_retry_task = models.OrderRetryTask() order_retry_task = models.OrderRetryTask()
order_retry_task.order_id = order.id order_retry_task.order_id = order.id
order_retry_task.retry_task = "foobar" order_retry_task.retry_task = "foobar"
@ -291,8 +293,7 @@ class WhenCreatingOrderRetryTask(utils.BaseTestCase):
class WhenCreatingNewCertificateAuthority(utils.BaseTestCase): class WhenCreatingNewCertificateAuthority(utils.BaseTestCase):
def setUp(self): def setUp(self):
super(WhenCreatingNewCertificateAuthority, self).setUp() super(WhenCreatingNewCertificateAuthority, self).setUp()
expiration = (datetime.datetime.utcnow() + expiration = timeutils.utcnow() + datetime.timedelta(minutes=10)
datetime.timedelta(minutes=10))
self.parsed_ca = {'plugin_name': 'dogtag_plugin', self.parsed_ca = {'plugin_name': 'dogtag_plugin',
'plugin_ca_id': 'ca_master', 'plugin_ca_id': 'ca_master',
'expiration': expiration.isoformat(), 'expiration': expiration.isoformat(),
@ -324,8 +325,7 @@ class WhenCreatingNewCertificateAuthority(utils.BaseTestCase):
class WhenCreatingNewProjectCertificateAuthority(utils.BaseTestCase): class WhenCreatingNewProjectCertificateAuthority(utils.BaseTestCase):
def setUp(self): def setUp(self):
super(WhenCreatingNewProjectCertificateAuthority, self).setUp() super(WhenCreatingNewProjectCertificateAuthority, self).setUp()
expiration = (datetime.datetime.utcnow() + expiration = timeutils.utcnow() + datetime.timedelta(minutes=10)
datetime.timedelta(minutes=10))
self.parsed_ca = {'plugin_name': 'dogtag_plugin', self.parsed_ca = {'plugin_name': 'dogtag_plugin',
'plugin_ca_id': 'ca_master', 'plugin_ca_id': 'ca_master',
'expiration': expiration.isoformat(), 'expiration': expiration.isoformat(),
@ -348,8 +348,7 @@ class WhenCreatingNewProjectCertificateAuthority(utils.BaseTestCase):
class WhenCreatingNewPreferredCertificateAuthority(utils.BaseTestCase): class WhenCreatingNewPreferredCertificateAuthority(utils.BaseTestCase):
def setUp(self): def setUp(self):
super(WhenCreatingNewPreferredCertificateAuthority, self).setUp() super(WhenCreatingNewPreferredCertificateAuthority, self).setUp()
expiration = (datetime.datetime.utcnow() + expiration = timeutils.utcnow() + datetime.timedelta(minutes=10)
datetime.timedelta(minutes=10))
self.parsed_ca = {'plugin_name': 'dogtag_plugin', self.parsed_ca = {'plugin_name': 'dogtag_plugin',
'plugin_ca_id': 'ca_master', 'plugin_ca_id': 'ca_master',
'expiration': expiration.isoformat(), 'expiration': expiration.isoformat(),

View File

@ -11,11 +11,11 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import datetime
import time import time
from unittest import mock from unittest import mock
import eventlet import eventlet
from oslo_utils import timeutils
from barbican.model import models from barbican.model import models
from barbican.model import repositories from barbican.model import repositories
@ -136,7 +136,7 @@ class WhenRunningPeriodicServerRetryLogic(database_utils.RepositoryTestCase):
retry = models.OrderRetryTask() retry = models.OrderRetryTask()
retry.order_id = order.id retry.order_id = order.id
retry.retry_at = datetime.datetime.utcnow() retry.retry_at = timeutils.utcnow()
retry.retry_task = task retry.retry_task = task
retry.retry_args = args retry.retry_args = args
retry.retry_kwargs = kwargs retry.retry_kwargs = kwargs

View File

@ -15,6 +15,8 @@
import datetime import datetime
from unittest import mock from unittest import mock
from oslo_utils import timeutils
from barbican.model import models from barbican.model import models
from barbican.model import repositories from barbican.model import repositories
from barbican.queue import server from barbican.queue import server
@ -189,10 +191,8 @@ class WhenCallingScheduleOrderRetryTasks(database_utils.RepositoryTestCase):
self.args = ['args-foo', 'args-bar'] self.args = ['args-foo', 'args-bar']
self.kwargs = {'order_id': self.order.id, 'foo': 1, 'bar': 2} self.kwargs = {'order_id': self.order.id, 'foo': 1, 'bar': 2}
self.date_to_retry_at = ( self.date_to_retry_at = timeutils.utcnow() + datetime.timedelta(
datetime.datetime.utcnow() + datetime.timedelta( milliseconds=self.result.retry_msec)
milliseconds=self.result.retry_msec)
)
def test_should_not_schedule_task_due_to_no_result(self): def test_should_not_schedule_task_due_to_no_result(self):
retry_rpc_method = server.schedule_order_retry_tasks(None, None, None) retry_rpc_method = server.schedule_order_retry_tasks(None, None, None)

View File

@ -14,12 +14,13 @@
# limitations under the License. # limitations under the License.
import datetime import datetime
from oslo_serialization import base64 as oslo_base64
from oslo_serialization import jsonutils as json
import sys import sys
import testtools
import time import time
from oslo_serialization import base64 as oslo_base64
from oslo_serialization import jsonutils as json
from oslo_utils import timeutils
import testtools
from testtools import testcase from testtools import testcase
from barbican.plugin.interface import secret_store as ss from barbican.plugin.interface import secret_store as ss
@ -1151,10 +1152,8 @@ class ListingSecretsTestCase(base.TestCase):
}) })
@testcase.attr('positive') @testcase.attr('positive')
def test_secret_list_with_date_filter(self, date_type): def test_secret_list_with_date_filter(self, date_type):
expiration_1 = str( expiration_1 = str(timeutils.utcnow() + datetime.timedelta(days=3))
datetime.datetime.utcnow() + datetime.timedelta(days=3)) expiration_2 = str(timeutils.utcnow() + datetime.timedelta(days=5))
expiration_2 = str(
datetime.datetime.utcnow() + datetime.timedelta(days=5))
two_phase_model = secret_models.SecretModel(expiration=expiration_1) two_phase_model = secret_models.SecretModel(expiration=expiration_1)
resp, secret_ref_1 = self.behaviors.create_secret(two_phase_model) resp, secret_ref_1 = self.behaviors.create_secret(two_phase_model)

View File

@ -18,7 +18,7 @@ oslo.policy>=3.11.0 # Apache-2.0
oslo.serialization!=2.19.1,>=2.18.0 # Apache-2.0 oslo.serialization!=2.19.1,>=2.18.0 # Apache-2.0
oslo.service!=1.28.1,>=1.24.0 # Apache-2.0 oslo.service!=1.28.1,>=1.24.0 # Apache-2.0
oslo.upgradecheck>=1.3.0 # Apache-2.0 oslo.upgradecheck>=1.3.0 # Apache-2.0
oslo.utils>=3.33.0 # Apache-2.0 oslo.utils>=7.0.0 # Apache-2.0
oslo.versionedobjects>=1.31.2 # Apache-2.0 oslo.versionedobjects>=1.31.2 # Apache-2.0
Paste>=2.0.2 # MIT Paste>=2.0.2 # MIT
PasteDeploy>=1.5.0 # MIT PasteDeploy>=1.5.0 # MIT