openstack
/
nova
Code Issues Proposed changes

Use TimeFixture from oslo_utils to override time in tests 

Instead of calling timeutils.set_time_override() directly where one
also has to remember to call timeutils.clear_time_override() when done
or potentially interfere with other tests, lets use the provided
fixture in oslo timeutils that handle all of this for us.

Also replace a few occurances where timeutils.utcnow() was mocked just
to provide some consistency in how time overrides are done.

Change-Id: Ice5bdb3fa3dc57ba8bafba702f2070fba7457ea7
This commit is contained in:
Hans Lindgren 2015-10-20 08:55:40 +02:00 committed by Sean Dague
parent 8dfaa494f5
commit 865063744f
16 changed files with 112 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
nova/tests/unit/api/ec2/test_middleware.py
View File

@ -17,7 +17,7 @@
from lxml import etree
import mock
from oslo_config import cfg
from oslo_utils import timeutils
from oslo_utils import fixture as utils_fixture
import requests
from six.moves import range
import webob
@ -45,13 +45,9 @@ class LockoutTestCase(test.NoDBTestCase):
"""Test case for the Lockout middleware."""
def setUp(self):
super(LockoutTestCase, self).setUp()
timeutils.set_time_override()
self.time_fixture = self.useFixture(utils_fixture.TimeFixture())
self.lockout = ec2.Lockout(conditional_forbid)
def tearDown(self):
timeutils.clear_time_override()
super(LockoutTestCase, self).tearDown()
def _send_bad_attempts(self, access_key, num_attempts=1):
"""Fail x."""
for i in range(num_attempts):
@ -70,21 +66,21 @@ class LockoutTestCase(test.NoDBTestCase):
def test_timeout(self):
self._send_bad_attempts('test', CONF.lockout_attempts)
self.assertTrue(self._is_locked_out('test'))
timeutils.advance_time_seconds(CONF.lockout_minutes * 60)
self.time_fixture.advance_time_seconds(CONF.lockout_minutes * 60)
self.assertFalse(self._is_locked_out('test'))
def test_multiple_keys(self):
self._send_bad_attempts('test1', CONF.lockout_attempts)
self.assertTrue(self._is_locked_out('test1'))
self.assertFalse(self._is_locked_out('test2'))
timeutils.advance_time_seconds(CONF.lockout_minutes * 60)
self.time_fixture.advance_time_seconds(CONF.lockout_minutes * 60)
self.assertFalse(self._is_locked_out('test1'))
self.assertFalse(self._is_locked_out('test2'))
def test_window_timeout(self):
self._send_bad_attempts('test', CONF.lockout_attempts - 1)
self.assertFalse(self._is_locked_out('test'))
timeutils.advance_time_seconds(CONF.lockout_window * 60)
self.time_fixture.advance_time_seconds(CONF.lockout_window * 60)
self._send_bad_attempts('test', CONF.lockout_attempts - 1)
self.assertFalse(self._is_locked_out('test'))

9
nova/tests/unit/api/openstack/compute/test_instance_usage_audit_log.py
View File

@ -15,7 +15,7 @@
import datetime
from oslo_utils import timeutils
from oslo_utils import fixture as utils_fixture
from nova.api.openstack.compute import instance_usage_audit_log as v21_ial
from nova.api.openstack.compute.legacy_v2.contrib \
@ -115,7 +115,8 @@ class InstanceUsageAuditLogTestV21(test.NoDBTestCase):
def setUp(self):
super(InstanceUsageAuditLogTestV21, self).setUp()
self.context = context.get_admin_context()
timeutils.set_time_override(datetime.datetime(2012, 7, 5, 10, 0, 0))
self.useFixture(
utils_fixture.TimeFixture(datetime.datetime(2012, 7, 5, 10, 0, 0)))
self._set_up_controller()
self.host_api = self.controller.host_api
@ -135,10 +136,6 @@ class InstanceUsageAuditLogTestV21(test.NoDBTestCase):
def _set_up_controller(self):
self.controller = v21_ial.InstanceUsageAuditLogController()
def tearDown(self):
super(InstanceUsageAuditLogTestV21, self).tearDown()
timeutils.clear_time_override()
def test_index(self):
result = self.controller.index(self.req)
self.assertIn('instance_usage_audit_logs', result)

4
nova/tests/unit/api/openstack/compute/test_server_usage.py
View File

@ -16,6 +16,7 @@
import datetime
from oslo_serialization import jsonutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from nova import compute
@ -102,8 +103,7 @@ class ServerUsageTestV21(test.TestCase):
res = self._make_request(url)
self.assertEqual(res.status_int, 200)
now = timeutils.utcnow()
timeutils.set_time_override(now)
self.useFixture(utils_fixture.TimeFixture())
self.assertServerUsage(self._get_server(res.body),
launched_at=DATE1,
terminated_at=DATE2)

7
nova/tests/unit/cells/test_cells_weights.py
View File

@ -20,6 +20,7 @@ Cells with higher weights should be given priority for new builds.
import datetime
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from nova.cells import state
@ -179,16 +180,12 @@ class MuteWeigherTestClass(_WeigherTestClass):
group='cells')
self.now = timeutils.utcnow()
timeutils.set_time_override(self.now)
self.useFixture(utils_fixture.TimeFixture(self.now))
self.cells = _get_fake_cells()
for cell in self.cells:
cell.last_seen = self.now
def tearDown(self):
super(MuteWeigherTestClass, self).tearDown()
timeutils.clear_time_override()
def test_non_mute(self):
weight_properties = {}
weighed_cells = self._get_weighed_cells(self.cells, weight_properties)

22
nova/tests/unit/compute/test_compute.py
View File

@ -33,6 +33,7 @@ from oslo_config import cfg
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_serialization import jsonutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import importutils
from oslo_utils import timeutils
from oslo_utils import units
@ -256,7 +257,6 @@ class BaseTestCase(test.TestCase):
self.rt = self.compute._get_resource_tracker(NODENAME)
def tearDown(self):
timeutils.clear_time_override()
ctxt = context.get_admin_context()
fake_image.FakeImageService_reset()
instances = db.instance_get_all(ctxt)
@ -2622,13 +2622,13 @@ class ComputeTestCase(BaseTestCase):
# Ensure instance can be rebuilt.
old_time = datetime.datetime(2012, 4, 1)
cur_time = datetime.datetime(2012, 12, 21, 12, 21)
timeutils.set_time_override(old_time)
time_fixture = self.useFixture(utils_fixture.TimeFixture(old_time))
instance = self._create_fake_instance_obj()
image_ref = instance['image_ref']
self.compute.build_and_run_instance(self.context, instance, {}, {}, {},
block_device_mapping=[])
timeutils.set_time_override(cur_time)
time_fixture.advance_time_delta(cur_time - old_time)
db.instance_update(self.context, instance['uuid'],
{"task_state": task_states.REBUILDING})
self.compute.rebuild_instance(self.context, instance,
@ -3924,13 +3924,13 @@ class ComputeTestCase(BaseTestCase):
old_time = datetime.datetime(2012, 4, 1)
cur_time = datetime.datetime(2012, 12, 21, 12, 21)
timeutils.set_time_override(old_time)
time_fixture = self.useFixture(utils_fixture.TimeFixture(old_time))
instance = self._create_fake_instance_obj()
self.compute.build_and_run_instance(self.context, instance, {}, {}, {},
block_device_mapping=[])
fake_notifier.NOTIFICATIONS = []
timeutils.set_time_override(cur_time)
time_fixture.advance_time_delta(cur_time - old_time)
self.compute.terminate_instance(self.context, instance, [], [])
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 4)
@ -4810,11 +4810,11 @@ class ComputeTestCase(BaseTestCase):
# Ensure notifications on instance migrate/resize.
old_time = datetime.datetime(2012, 4, 1)
cur_time = datetime.datetime(2012, 12, 21, 12, 21)
timeutils.set_time_override(old_time)
time_fixture = self.useFixture(utils_fixture.TimeFixture(old_time))
inst_ref = self._create_fake_instance_obj()
self.compute.build_and_run_instance(self.context, inst_ref, {}, {}, {},
block_device_mapping=[])
timeutils.set_time_override(cur_time)
time_fixture.advance_time_delta(cur_time - old_time)
fake_notifier.NOTIFICATIONS = []
instance = db.instance_get_by_uuid(self.context, inst_ref['uuid'])
@ -4878,7 +4878,7 @@ class ComputeTestCase(BaseTestCase):
# Ensure notifications on instance migrate/resize.
old_time = datetime.datetime(2012, 4, 1)
cur_time = datetime.datetime(2012, 12, 21, 12, 21)
timeutils.set_time_override(old_time)
time_fixture = self.useFixture(utils_fixture.TimeFixture(old_time))
instance = self._create_fake_instance_obj()
new_type = flavors.get_flavor_by_name('m1.small')
new_type_id = new_type['id']
@ -4903,7 +4903,7 @@ class ComputeTestCase(BaseTestCase):
self.compute.resize_instance(self.context, instance=instance,
migration=migration, image={}, instance_type=new_type,
reservations=[], clean_shutdown=True)
timeutils.set_time_override(cur_time)
time_fixture.advance_time_delta(cur_time - old_time)
fake_notifier.NOTIFICATIONS = []
self.compute.finish_resize(self.context,
@ -4937,12 +4937,12 @@ class ComputeTestCase(BaseTestCase):
# Ensure notifications on instance migrate/resize.
old_time = datetime.datetime(2012, 4, 1)
cur_time = datetime.datetime(2012, 12, 21, 12, 21)
timeutils.set_time_override(old_time)
time_fixture = self.useFixture(utils_fixture.TimeFixture(old_time))
instance = self._create_fake_instance_obj()
self.compute.build_and_run_instance(self.context, instance, {}, {}, {},
block_device_mapping=[])
timeutils.set_time_override(cur_time)
time_fixture.advance_time_delta(cur_time - old_time)
fake_notifier.NOTIFICATIONS = []
instance.host = 'foo'

5
nova/tests/unit/compute/test_compute_api.py
View File

@ -21,6 +21,7 @@ import mock
from mox3 import mox
from oslo_policy import policy as oslo_policy
from oslo_serialization import jsonutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from oslo_utils import uuidutils
@ -756,7 +757,7 @@ class _ComputeAPIUnitTestMixIn(object):
'ram': -inst.memory_mb}
delete_time = datetime.datetime(1955, 11, 5, 9, 30,
tzinfo=iso8601.iso8601.Utc())
timeutils.set_time_override(delete_time)
self.useFixture(utils_fixture.TimeFixture(delete_time))
task_state = (delete_type == 'soft_delete' and
task_states.SOFT_DELETING or task_states.DELETING)
updates = {'progress': 0, 'task_state': task_state}
@ -1093,7 +1094,7 @@ class _ComputeAPIUnitTestMixIn(object):
self.mox.StubOutWithMock(inst, 'save')
delete_time = datetime.datetime(1955, 11, 5)
timeutils.set_time_override(delete_time)
self.useFixture(utils_fixture.TimeFixture(delete_time))
db.block_device_mapping_get_all_by_instance(
self.context, inst.uuid, use_slave=False).AndReturn([])

15
nova/tests/unit/compute/test_shelve.py
View File

@ -13,6 +13,7 @@
import mock
from mox3 import mox
from oslo_config import cfg
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from nova.compute import claims
@ -48,8 +49,7 @@ class ShelveComputeManagerTestCase(test_compute.BaseTestCase):
instance = self._create_fake_instance_obj(params={'host': host})
image_id = 'fake_image_id'
host = 'fake-mini'
cur_time = timeutils.utcnow()
timeutils.set_time_override(cur_time)
self.useFixture(utils_fixture.TimeFixture())
instance.task_state = task_states.SHELVING
instance.save()
@ -141,8 +141,7 @@ class ShelveComputeManagerTestCase(test_compute.BaseTestCase):
instance = self._create_fake_instance_obj(params={'host': host})
instance.task_state = task_states.SHELVING
instance.save()
cur_time = timeutils.utcnow()
timeutils.set_time_override(cur_time)
self.useFixture(utils_fixture.TimeFixture())
self.mox.StubOutWithMock(self.compute, '_notify_about_instance_usage')
self.mox.StubOutWithMock(self.compute.driver, 'power_off')
@ -353,8 +352,8 @@ class ShelveComputeManagerTestCase(test_compute.BaseTestCase):
mock_older.return_value = False
self.flags(shelved_offload_time=1)
shelved_time = timeutils.utcnow()
timeutils.set_time_override(shelved_time)
timeutils.advance_time_seconds(CONF.shelved_offload_time - 1)
time_fixture = self.useFixture(utils_fixture.TimeFixture(shelved_time))
time_fixture.advance_time_seconds(CONF.shelved_offload_time - 1)
instance = self._create_fake_instance_obj()
instance.vm_state = vm_states.SHELVED
instance.task_state = None
@ -371,8 +370,8 @@ class ShelveComputeManagerTestCase(test_compute.BaseTestCase):
def test_shelved_poll_timedout(self):
self.flags(shelved_offload_time=1)
shelved_time = timeutils.utcnow()
timeutils.set_time_override(shelved_time)
timeutils.advance_time_seconds(CONF.shelved_offload_time + 10)
time_fixture = self.useFixture(utils_fixture.TimeFixture(shelved_time))
time_fixture.advance_time_seconds(CONF.shelved_offload_time + 1)
instance = self._create_fake_instance_obj()
instance.vm_state = vm_states.SHELVED
instance.task_state = None

11
nova/tests/unit/db/test_db_api.py
View File

@ -32,6 +32,7 @@ from oslo_db.sqlalchemy import test_base
from oslo_db.sqlalchemy import update_match
from oslo_db.sqlalchemy import utils as sqlalchemyutils
from oslo_serialization import jsonutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from oslo_utils import uuidutils
import six
@ -5547,7 +5548,7 @@ class VolumeUsageDBApiTestCase(test.TestCase):
def test_vol_usage_update_no_totals_update(self):
ctxt = context.get_admin_context()
now = timeutils.utcnow()
timeutils.set_time_override(now)
self.useFixture(utils_fixture.TimeFixture(now))
start_time = now - datetime.timedelta(seconds=10)
expected_vol_usages = {
@ -5619,7 +5620,7 @@ class VolumeUsageDBApiTestCase(test.TestCase):
now2 = now + datetime.timedelta(minutes=2)
now3 = now + datetime.timedelta(minutes=3)
timeutils.set_time_override(now)
time_fixture = self.useFixture(utils_fixture.TimeFixture(now))
db.vol_usage_update(ctxt, u'1', rd_req=100, rd_bytes=200,
wr_req=300, wr_bytes=400,
instance_id='fake-instance-uuid',
@ -5630,7 +5631,7 @@ class VolumeUsageDBApiTestCase(test.TestCase):
self.assertEqual(current_usage['tot_reads'], 0)
self.assertEqual(current_usage['curr_reads'], 100)
timeutils.set_time_override(now1)
time_fixture.advance_time_delta(now1 - now)
db.vol_usage_update(ctxt, u'1', rd_req=200, rd_bytes=300,
wr_req=400, wr_bytes=500,
instance_id='fake-instance-uuid',
@ -5642,7 +5643,7 @@ class VolumeUsageDBApiTestCase(test.TestCase):
self.assertEqual(current_usage['tot_reads'], 200)
self.assertEqual(current_usage['curr_reads'], 0)
timeutils.set_time_override(now2)
time_fixture.advance_time_delta(now2 - now1)
db.vol_usage_update(ctxt, u'1', rd_req=300, rd_bytes=400,
wr_req=500, wr_bytes=600,
instance_id='fake-instance-uuid',
@ -5653,7 +5654,7 @@ class VolumeUsageDBApiTestCase(test.TestCase):
self.assertEqual(current_usage['tot_reads'], 200)
self.assertEqual(current_usage['curr_reads'], 300)
timeutils.set_time_override(now3)
time_fixture.advance_time_delta(now3 - now2)
db.vol_usage_update(ctxt, u'1', rd_req=400, rd_bytes=500,
wr_req=600, wr_bytes=700,
instance_id='fake-instance-uuid',

37
nova/tests/unit/objects/test_instance_action.py
View File

@ -15,6 +15,7 @@
import traceback
import mock
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
import six
@ -78,7 +79,7 @@ class _TestInstanceActionObject(object):
self.context.timestamp)
def test_pack_action_finish(self):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
values = instance_action.InstanceAction.pack_action_finish(
self.context, 'fake-uuid')
self.assertEqual(values['request_id'], self.context.request_id)
@ -111,7 +112,7 @@ class _TestInstanceActionObject(object):
@mock.patch.object(db, 'action_finish')
def test_action_finish(self, mock_finish):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceAction
expected_packed_values = test_class.pack_action_finish(
self.context, 'fake-uuid')
@ -124,7 +125,7 @@ class _TestInstanceActionObject(object):
@mock.patch.object(db, 'action_finish')
def test_action_finish_no_result(self, mock_finish):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceAction
expected_packed_values = test_class.pack_action_finish(
self.context, 'fake-uuid')
@ -138,7 +139,7 @@ class _TestInstanceActionObject(object):
@mock.patch.object(db, 'action_finish')
@mock.patch.object(db, 'action_start')
def test_finish(self, mock_start, mock_finish):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
expected_packed_action_start = {
'request_id': self.context.request_id,
'user_id': self.context.user_id,
@ -197,7 +198,7 @@ class _TestInstanceActionEventObject(object):
@mock.patch.object(db, 'action_event_start')
def test_event_start(self, mock_start):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_start(
self.context, 'fake-uuid', 'fake-event')
@ -210,7 +211,7 @@ class _TestInstanceActionEventObject(object):
@mock.patch.object(db, 'action_event_start')
def test_event_start_no_result(self, mock_start):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_start(
self.context, 'fake-uuid', 'fake-event')
@ -223,11 +224,11 @@ class _TestInstanceActionEventObject(object):
@mock.patch.object(db, 'action_event_finish')
def test_event_finish(self, mock_finish):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_finish(
self.context, 'fake-uuid', 'fake-event')
expected_packed_values['finish_time'] = timeutils.utcnow()
expected_packed_values['finish_time'] = NOW
mock_finish.return_value = fake_event
event = instance_action.InstanceActionEvent.event_finish(
self.context, 'fake-uuid', 'fake-event', want_result=True)
@ -237,11 +238,11 @@ class _TestInstanceActionEventObject(object):
@mock.patch.object(db, 'action_event_finish')
def test_event_finish_no_result(self, mock_finish):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_finish(
self.context, 'fake-uuid', 'fake-event')
expected_packed_values['finish_time'] = timeutils.utcnow()
expected_packed_values['finish_time'] = NOW
mock_finish.return_value = fake_event
event = instance_action.InstanceActionEvent.event_finish(
self.context, 'fake-uuid', 'fake-event', want_result=False)
@ -252,11 +253,11 @@ class _TestInstanceActionEventObject(object):
@mock.patch.object(traceback, 'format_tb')
@mock.patch.object(db, 'action_event_finish')
def test_event_finish_with_failure(self, mock_finish, mock_tb):
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_finish(
self.context, 'fake-uuid', 'fake-event', 'val', 'fake-tb')
expected_packed_values['finish_time'] = timeutils.utcnow()
expected_packed_values['finish_time'] = NOW
mock_finish.return_value = fake_event
event = test_class.event_finish_with_failure(
@ -271,11 +272,11 @@ class _TestInstanceActionEventObject(object):
def test_event_finish_with_failure_legacy(self, mock_finish, mock_tb):
# Tests that exc_tb is serialized when it's not a string type.
mock_tb.return_value = 'fake-tb'
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_finish(
self.context, 'fake-uuid', 'fake-event', 'val', 'fake-tb')
expected_packed_values['finish_time'] = timeutils.utcnow()
expected_packed_values['finish_time'] = NOW
mock_finish.return_value = fake_event
fake_tb = mock.sentinel.fake_tb
@ -290,12 +291,12 @@ class _TestInstanceActionEventObject(object):
@mock.patch.object(db, 'action_event_finish')
def test_event_finish_with_failure_legacy_unicode(self, mock_finish):
# Tests that traceback.format_tb is not called when exc_tb is unicode.
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_finish(
self.context, 'fake-uuid', 'fake-event', 'val',
six.text_type('fake-tb'))
expected_packed_values['finish_time'] = timeutils.utcnow()
expected_packed_values['finish_time'] = NOW
mock_finish.return_value = fake_event
event = test_class.event_finish_with_failure(
@ -311,11 +312,11 @@ class _TestInstanceActionEventObject(object):
# Tests that traceback.format_tb is not called when exc_tb is a str
# and want_result is False, so no event should come back.
mock_tb.return_value = 'fake-tb'
timeutils.set_time_override(override_time=NOW)
self.useFixture(utils_fixture.TimeFixture(NOW))
test_class = instance_action.InstanceActionEvent
expected_packed_values = test_class.pack_action_event_finish(
self.context, 'fake-uuid', 'fake-event', 'val', 'fake-tb')
expected_packed_values['finish_time'] = timeutils.utcnow()
expected_packed_values['finish_time'] = NOW
mock_finish.return_value = fake_event
event = test_class.event_finish_with_failure(

15
nova/tests/unit/scheduler/filters/test_trusted_filters.py
View File

@ -13,6 +13,7 @@
import mock
from oslo_config import cfg
from oslo_serialization import jsonutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
import requests
@ -187,14 +188,12 @@ class TestTrustedFilter(test.NoDBTestCase):
req_mock.reset_mock()
timeutils.set_time_override(timeutils.utcnow())
timeutils.advance_time_seconds(
time_fixture = self.useFixture(utils_fixture.TimeFixture())
time_fixture.advance_time_seconds(
CONF.trusted_computing.attestation_auth_timeout + 80)
self.filt_cls.host_passes(host, spec_obj)
self.assertTrue(req_mock.called)
timeutils.clear_time_override()
def test_trusted_filter_update_cache_timezone(self, req_mock):
oat_data = {"hosts": [{"host_name": "node1",
"trust_lvl": "untrusted",
@ -207,9 +206,9 @@ class TestTrustedFilter(test.NoDBTestCase):
extra_specs=extra_specs))
host = fakes.FakeHostState('host1', 'node1', {})
timeutils.set_time_override(
time_fixture = self.useFixture(utils_fixture.TimeFixture(
timeutils.normalize_time(
timeutils.parse_isotime("2012-09-09T09:10:40Z")))
timeutils.parse_isotime("2012-09-09T09:10:40Z"))))
self.filt_cls.host_passes(host, spec_obj) # Fill the caches
@ -218,13 +217,11 @@ class TestTrustedFilter(test.NoDBTestCase):
self.assertFalse(req_mock.called)
req_mock.reset_mock()
timeutils.advance_time_seconds(
time_fixture.advance_time_seconds(
CONF.trusted_computing.attestation_auth_timeout - 10)
self.filt_cls.host_passes(host, spec_obj)
self.assertFalse(req_mock.called)
timeutils.clear_time_override()
def test_trusted_filter_combine_hosts(self, req_mock):
fake_compute_nodes = [
objects.ComputeNode(hypervisor_hostname='node1'),

45
nova/tests/unit/servicegroup/test_db_servicegroup.py
View File

@ -12,10 +12,11 @@
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import mock
from oslo_db import exception as db_exception
import oslo_messaging as messaging
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from nova import objects
from nova import servicegroup
@ -31,42 +32,36 @@ class DBServiceGroupTestCase(test.NoDBTestCase):
servicegroup_driver='db')
self.servicegroup_api = servicegroup.API()
@mock.patch('oslo_utils.timeutils.utcnow')
def test_is_up(self, now_mock):
service_ref = {
'host': 'fake-host',
'topic': 'compute',
}
fts_func = datetime.datetime.fromtimestamp
fake_now = 1000
def test_is_up(self):
now = timeutils.utcnow()
service = objects.Service(
host='fake-host',
topic='compute',
created_at=now,
updated_at=now,
last_seen_up=now,
forced_down=False,
)
time_fixture = self.useFixture(utils_fixture.TimeFixture(now))
# Up (equal)
now_mock.return_value = fts_func(fake_now)
service_ref['last_seen_up'] = fts_func(fake_now - self.down_time)
service_ref['updated_at'] = fts_func(fake_now - self.down_time)
service_ref['created_at'] = fts_func(fake_now - self.down_time)
result = self.servicegroup_api.service_is_up(service_ref)
result = self.servicegroup_api.service_is_up(service)
self.assertTrue(result)
# Up
service_ref['last_seen_up'] = fts_func(fake_now - self.down_time + 1)
service_ref['updated_at'] = fts_func(fake_now - self.down_time + 1)
service_ref['created_at'] = fts_func(fake_now - self.down_time + 1)
result = self.servicegroup_api.service_is_up(service_ref)
time_fixture.advance_time_seconds(self.down_time)
result = self.servicegroup_api.service_is_up(service)
self.assertTrue(result)
# Down
service_ref['last_seen_up'] = fts_func(fake_now - self.down_time - 3)
service_ref['updated_at'] = fts_func(fake_now - self.down_time - 3)
service_ref['created_at'] = fts_func(fake_now - self.down_time - 3)
result = self.servicegroup_api.service_is_up(service_ref)
time_fixture.advance_time_seconds(1)
result = self.servicegroup_api.service_is_up(service)
self.assertFalse(result)
# "last_seen_up" says down, "updated_at" says up.
# This can happen if we do a service disable/enable while it's down.
service_ref['updated_at'] = fts_func(fake_now - self.down_time + 1)
result = self.servicegroup_api.service_is_up(service_ref)
service.updated_at = timeutils.utcnow()
result = self.servicegroup_api.service_is_up(service)
self.assertFalse(result)
def test_join(self):

20
nova/tests/unit/test_utils.py
View File

@ -30,7 +30,7 @@ from oslo_config import cfg
from oslo_context import context as common_context
from oslo_context import fixture as context_fixture
from oslo_utils import encodeutils
from oslo_utils import timeutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import units
import six
@ -586,17 +586,13 @@ class AuditPeriodTest(test.NoDBTestCase):
def setUp(self):
super(AuditPeriodTest, self).setUp()
# a fairly random time to test with
self.test_time = datetime.datetime(second=23,
minute=12,
hour=8,
day=5,
month=3,
year=2012)
timeutils.set_time_override(override_time=self.test_time)
def tearDown(self):
timeutils.clear_time_override()
super(AuditPeriodTest, self).tearDown()
self.useFixture(utils_fixture.TimeFixture(
datetime.datetime(second=23,
minute=12,
hour=8,
day=5,
month=3,
year=2012)))
def test_hour(self):
begin, end = utils.last_completed_audit_period(unit='hour')

18
nova/tests/unit/virt/libvirt/test_driver.py
View File

@ -41,8 +41,8 @@ from oslo_serialization import jsonutils
from oslo_service import loopingcall
from oslo_utils import encodeutils
from oslo_utils import fileutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import importutils
from oslo_utils import timeutils
from oslo_utils import units
from oslo_utils import uuidutils
from oslo_utils import versionutils
@ -10118,7 +10118,7 @@ class LibvirtConnTestCase(test.NoDBTestCase):
lt = datetime.datetime(2012, 11, 22, 12, 00, 00)
diags_time = datetime.datetime(2012, 11, 22, 12, 00, 10)
timeutils.set_time_override(diags_time)
self.useFixture(utils_fixture.TimeFixture(diags_time))
instance.launched_at = lt
actual = drvr.get_instance_diagnostics(instance)
@ -10229,7 +10229,7 @@ class LibvirtConnTestCase(test.NoDBTestCase):
lt = datetime.datetime(2012, 11, 22, 12, 00, 00)
diags_time = datetime.datetime(2012, 11, 22, 12, 00, 10)
timeutils.set_time_override(diags_time)
self.useFixture(utils_fixture.TimeFixture(diags_time))
instance.launched_at = lt
actual = drvr.get_instance_diagnostics(instance)
@ -10334,7 +10334,7 @@ class LibvirtConnTestCase(test.NoDBTestCase):
lt = datetime.datetime(2012, 11, 22, 12, 00, 00)
diags_time = datetime.datetime(2012, 11, 22, 12, 00, 10)
timeutils.set_time_override(diags_time)
self.useFixture(utils_fixture.TimeFixture(diags_time))
instance.launched_at = lt
actual = drvr.get_instance_diagnostics(instance)
@ -10448,7 +10448,7 @@ class LibvirtConnTestCase(test.NoDBTestCase):
lt = datetime.datetime(2012, 11, 22, 12, 00, 00)
diags_time = datetime.datetime(2012, 11, 22, 12, 00, 10)
timeutils.set_time_override(diags_time)
self.useFixture(utils_fixture.TimeFixture(diags_time))
instance.launched_at = lt
actual = drvr.get_instance_diagnostics(instance)
@ -10572,7 +10572,7 @@ class LibvirtConnTestCase(test.NoDBTestCase):
lt = datetime.datetime(2012, 11, 22, 12, 00, 00)
diags_time = datetime.datetime(2012, 11, 22, 12, 00, 10)
timeutils.set_time_override(diags_time)
self.useFixture(utils_fixture.TimeFixture(diags_time))
instance.launched_at = lt
actual = drvr.get_instance_diagnostics(instance)
@ -10610,10 +10610,8 @@ class LibvirtConnTestCase(test.NoDBTestCase):
'version': '1.0'}
self.assertEqual(expected, actual.serialize())
@mock.patch.object(timeutils, 'utcnow')
@mock.patch.object(host.Host, 'get_domain')
def test_diagnostic_full_with_multiple_interfaces(self, mock_get_domain,
mock_utcnow):
def test_diagnostic_full_with_multiple_interfaces(self, mock_get_domain):
xml = """
<domain type='kvm'>
<devices>
@ -10712,7 +10710,7 @@ class LibvirtConnTestCase(test.NoDBTestCase):
lt = datetime.datetime(2012, 11, 22, 12, 00, 00)
diags_time = datetime.datetime(2012, 11, 22, 12, 00, 10)
mock_utcnow.return_value = diags_time
self.useFixture(utils_fixture.TimeFixture(diags_time))
instance.launched_at = lt
actual = drvr.get_instance_diagnostics(instance)

4
nova/tests/unit/virt/vmwareapi/test_driver_api.py
View File

@ -26,7 +26,7 @@ from eventlet import greenthread
import mock
from mox3 import mox
from oslo_config import cfg
from oslo_utils import timeutils
from oslo_utils import fixture as utils_fixture
from oslo_utils import units
from oslo_utils import uuidutils
from oslo_vmware import exceptions as vexc
@ -1974,7 +1974,7 @@ class VMwareAPIVMTestCase(test.NoDBTestCase):
self.flags(remove_unused_original_minimum_age_seconds=aging_time)
self._image_aging_image_marked_for_deletion()
all_instances = []
timeutils.set_time_override(cur_time)
self.useFixture(utils_fixture.TimeFixture(cur_time))
self.conn.manage_image_cache(self.context, all_instances)
def test_image_aging_aged(self):

6
nova/tests/unit/virt/vmwareapi/test_imagecache.py
View File

@ -16,7 +16,7 @@ import datetime
import mock
from oslo_config import cfg
from oslo_utils import timeutils
from oslo_utils import fixture as utils_fixture
from oslo_vmware.objects import datastore as ds_obj
from oslo_vmware import vim_util as vutil
@ -99,7 +99,7 @@ class ImageCacheManagerTestCase(test.NoDBTestCase):
self.assertIsNone(ts)
def test_get_timestamp_filename(self):
timeutils.set_time_override(override_time=self._time)
self.useFixture(utils_fixture.TimeFixture(self._time))
fn = self._imagecache._get_timestamp_filename()
self.assertEqual(self._file_name, fn)
@ -217,7 +217,7 @@ class ImageCacheManagerTestCase(test.NoDBTestCase):
fake_timestamp_cleanup),
) as (_get_ds_browser, _get_timestamp, _mkdir, _file_delete,
_timestamp_cleanup):
timeutils.set_time_override(override_time=self._time)
self.useFixture(utils_fixture.TimeFixture(self._time))
datastore = ds_obj.Datastore(name='ds', ref='fake-ds-ref')
dc_info = ds_util.DcInfo(ref='dc_ref', name='name',
vmFolder='vmFolder')

11
nova/tests/unit/virt/xenapi/test_vm_utils.py
View File

@ -24,6 +24,7 @@ from oslo_concurrency import lockutils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_config import fixture as config_fixture
from oslo_utils import fixture as utils_fixture
from oslo_utils import timeutils
from oslo_utils import units
from oslo_utils import uuidutils
@ -528,16 +529,18 @@ class ResizeHelpersTestCase(VMUtilsTestBase):
"%(left)s bytes left to copy",
{"complete_pct": 50.0, "left": 1})
current = timeutils.utcnow()
timeutils.set_time_override(current)
timeutils.advance_time_seconds(vm_utils.PROGRESS_INTERVAL_SECONDS + 1)
time_fixture = self.useFixture(utils_fixture.TimeFixture(current))
time_fixture.advance_time_seconds(
vm_utils.PROGRESS_INTERVAL_SECONDS + 1)
self.mox.ReplayAll()
vm_utils._log_progress_if_required(1, current, 2)
def test_log_progress_if_not_required(self):
self.mox.StubOutWithMock(vm_utils.LOG, "debug")
current = timeutils.utcnow()
timeutils.set_time_override(current)
timeutils.advance_time_seconds(vm_utils.PROGRESS_INTERVAL_SECONDS - 1)
time_fixture = self.useFixture(utils_fixture.TimeFixture(current))
time_fixture.advance_time_seconds(
vm_utils.PROGRESS_INTERVAL_SECONDS - 1)
self.mox.ReplayAll()
vm_utils._log_progress_if_required(1, current, 2)