openstack/barbican
Code Issues Proposed changes

Merge "Replace deprecated datetime.utcnow()"

Browse Source
This commit is contained in:
Zuul 2024-11-21 04:12:36 +00:00 committed by Gerrit Code Review
commit 49c7ac543f
13 changed files with 44 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
barbican/queue/retry_scheduler.py
View File

@ -16,10 +16,10 @@
"""
Retry/scheduler classes and logic.
"""
import datetime
import random
from oslo_service import service
from oslo_utils import timeutils
from barbican.common import config
from barbican.common import utils
@ -118,7 +118,7 @@ class PeriodicServer(service.Service):
repositories.start()
try:
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)
finally:
repositories.clear()

3
barbican/queue/server.py
View File

@ -27,6 +27,7 @@ except ImportError:
newrelic_loaded = False
from oslo_service import service
from oslo_utils import timeutils
from barbican.common import utils
from barbican.model import models
@ -166,7 +167,7 @@ def schedule_order_retry_tasks(
LOG.debug(
'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)
retry_model = models.OrderRetryTask()

5
barbican/tests/cmd/test_db_cleanup.py
View File

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

3
barbican/tests/common/test_validators.py
View File

@ -17,6 +17,7 @@ import datetime
import unittest
from oslo_serialization import base64
from oslo_utils import timeutils
from barbican.common import exception as excep
from barbican.common import validators
@ -1203,7 +1204,7 @@ class WhenTestingKeyTypeOrderValidator(utils.BaseTestCase):
def test_should_pass_good_exp_meta_in_order_refs(self):
self.key_order_req['meta']['algorithm'] = 'AES'
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')
self.key_order_req['meta']['expiration'] = date_after_year_str
result = self.validator.validate(self.key_order_req)

5
barbican/tests/database_utils.py
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
break the DevStack functional test discovery process.
"""
import datetime
from oslo_utils import timeutils
import oslotest.base as oslotest
from sqlalchemy.engine import Engine
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_kwargs = retry_kwargs
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_task_repo = repositories.get_order_retry_tasks_repository()
order_retry_task_repo.create_from(order_retry, session)

14
barbican/tests/model/repositories/test_repositories_certificate_authorities.py
View File

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

6
barbican/tests/model/repositories/test_repositories_order_retry_tasks.py
View File

@ -15,6 +15,8 @@
import datetime
import time
from oslo_utils import timeutils
from barbican.common import config
from barbican.common import exception
from barbican.model import models
@ -27,7 +29,7 @@ class WhenTestingOrderRetryTaskRepository(database_utils.RepositoryTestCase):
def setUp(self):
super(WhenTestingOrderRetryTaskRepository, self).setUp()
self.date_time_now = datetime.datetime.utcnow()
self.date_time_now = timeutils.utcnow()
self.test_args = ['test', 'args']
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.
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,
suppress_exception=True
)

3
barbican/tests/model/repositories/test_repositories_secrets.py
View File

@ -13,6 +13,7 @@
import datetime
import fixtures
from oslo_utils import timeutils
import testtools
from barbican.common import exception
@ -222,7 +223,7 @@ class WhenTestingSecretRepository(database_utils.RepositoryTestCase):
self.assertEqual(1, count)
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)
yesterday = current_time - datetime.timedelta(days=1)

17
barbican/tests/model/test_models.py
View File

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

4
barbican/tests/queue/test_retry_scheduler.py
View File

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

8
barbican/tests/queue/test_server.py
View File

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

13
functionaltests/api/v1/functional/test_secrets.py
View File

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

2
requirements.txt
View File

@ -18,7 +18,7 @@ oslo.policy>=4.5.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.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
Paste>=2.0.2 # MIT
PasteDeploy>=1.5.0 # MIT