010fe3b102
The move from iso8601===0.1.11 to iso8601===0.1.12 broke unit
tests in oslo.utils.
iso8601 used to do:
from datetime import datetime
But now they call datetime.datetime():
import datetime
datetime.datetime()
Unfortunately the unit tests that mocked datetime.datetime() are now
mocking the one in iso8601. This causes a failure in the unit tests.
Fix this by using the 'wraps' argument to mock. So that the calls will
get passed through to datetime.datetime. Also changed to using the
decorator style of mock.
In addition Python 3 unit tests were broken due to changing how the
UTC time zone is represented from 'UTC' to 'UTC+00:00'.
Closes-Bug: #1747575
Closes-Bug: #1744160
Change-Id: Ia80ffb5e571cc5366bef2bc1a32c457a3c16843d
626 lines
24 KiB
Python
626 lines
24 KiB
Python
# Copyright 2011 OpenStack Foundation.
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import calendar
|
|
import datetime
|
|
import logging
|
|
import time
|
|
|
|
import iso8601
|
|
import mock
|
|
from oslotest import base as test_base
|
|
from testtools import matchers
|
|
|
|
from oslo_utils import timeutils
|
|
|
|
|
|
def monotonic_iter(start=0, incr=0.05):
|
|
while True:
|
|
yield start
|
|
start += incr
|
|
|
|
|
|
class TimeUtilsTest(test_base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TimeUtilsTest, self).setUp()
|
|
self.skynet_self_aware_time_str = '1997-08-29T06:14:00Z'
|
|
self.skynet_self_aware_time_ms_str = '1997-08-29T06:14:00.000123Z'
|
|
self.skynet_self_aware_time = datetime.datetime(1997, 8, 29, 6, 14, 0)
|
|
self.skynet_self_aware_ms_time = datetime.datetime(1997, 8, 29, 6, 14,
|
|
0, 123)
|
|
self.one_minute_before = datetime.datetime(1997, 8, 29, 6, 13, 0)
|
|
self.one_minute_after = datetime.datetime(1997, 8, 29, 6, 15, 0)
|
|
self.skynet_self_aware_time_perfect_str = '1997-08-29T06:14:00.000000'
|
|
self.skynet_self_aware_time_perfect = datetime.datetime(1997, 8, 29,
|
|
6, 14, 0)
|
|
self.addCleanup(timeutils.clear_time_override)
|
|
|
|
def test_isotime(self):
|
|
with mock.patch('datetime.datetime') as datetime_mock:
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_time
|
|
dt = timeutils.isotime()
|
|
self.assertEqual(dt, self.skynet_self_aware_time_str)
|
|
|
|
def test_isotimei_micro_second_precision(self):
|
|
with mock.patch('datetime.datetime') as datetime_mock:
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_ms_time
|
|
dt = timeutils.isotime(subsecond=True)
|
|
self.assertEqual(dt, self.skynet_self_aware_time_ms_str)
|
|
|
|
def test_parse_isotime(self):
|
|
expect = timeutils.parse_isotime(self.skynet_self_aware_time_str)
|
|
skynet_self_aware_time_utc = self.skynet_self_aware_time.replace(
|
|
tzinfo=iso8601.iso8601.UTC)
|
|
self.assertEqual(skynet_self_aware_time_utc, expect)
|
|
|
|
def test_parse_isotime_micro_second_precision(self):
|
|
expect = timeutils.parse_isotime(self.skynet_self_aware_time_ms_str)
|
|
skynet_self_aware_time_ms_utc = self.skynet_self_aware_ms_time.replace(
|
|
tzinfo=iso8601.iso8601.UTC)
|
|
self.assertEqual(skynet_self_aware_time_ms_utc, expect)
|
|
|
|
def test_strtime(self):
|
|
expect = timeutils.strtime(self.skynet_self_aware_time_perfect)
|
|
self.assertEqual(self.skynet_self_aware_time_perfect_str, expect)
|
|
|
|
def test_parse_strtime(self):
|
|
perfect_time_format = self.skynet_self_aware_time_perfect_str
|
|
expect = timeutils.parse_strtime(perfect_time_format)
|
|
self.assertEqual(self.skynet_self_aware_time_perfect, expect)
|
|
|
|
def test_strtime_and_back(self):
|
|
orig_t = datetime.datetime(1997, 8, 29, 6, 14, 0)
|
|
s = timeutils.strtime(orig_t)
|
|
t = timeutils.parse_strtime(s)
|
|
self.assertEqual(orig_t, t)
|
|
|
|
@mock.patch('datetime.datetime', wraps=datetime.datetime)
|
|
def _test_is_older_than(self, fn, datetime_mock):
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_time
|
|
expect_true = timeutils.is_older_than(fn(self.one_minute_before),
|
|
59)
|
|
self.assertTrue(expect_true)
|
|
expect_false = timeutils.is_older_than(fn(self.one_minute_before),
|
|
60)
|
|
self.assertFalse(expect_false)
|
|
expect_false = timeutils.is_older_than(fn(self.one_minute_before),
|
|
61)
|
|
self.assertFalse(expect_false)
|
|
|
|
def test_is_older_than_datetime(self):
|
|
self._test_is_older_than(lambda x: x)
|
|
|
|
def test_is_older_than_str(self):
|
|
self._test_is_older_than(timeutils.strtime)
|
|
|
|
def test_is_older_than_aware(self):
|
|
"""Tests sending is_older_than an 'aware' datetime."""
|
|
self._test_is_older_than(lambda x: x.replace(
|
|
tzinfo=iso8601.iso8601.UTC))
|
|
|
|
def test_is_older_than_aware_no_utc(self):
|
|
self._test_is_older_than(lambda x: x.replace(
|
|
tzinfo=iso8601.iso8601.FixedOffset(1, 0, 'foo')).replace(
|
|
hour=7))
|
|
|
|
@mock.patch('datetime.datetime', wraps=datetime.datetime)
|
|
def _test_is_newer_than(self, fn, datetime_mock):
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_time
|
|
expect_true = timeutils.is_newer_than(fn(self.one_minute_after),
|
|
59)
|
|
self.assertTrue(expect_true)
|
|
expect_false = timeutils.is_newer_than(fn(self.one_minute_after),
|
|
60)
|
|
self.assertFalse(expect_false)
|
|
expect_false = timeutils.is_newer_than(fn(self.one_minute_after),
|
|
61)
|
|
self.assertFalse(expect_false)
|
|
|
|
def test_is_newer_than_datetime(self):
|
|
self._test_is_newer_than(lambda x: x)
|
|
|
|
def test_is_newer_than_str(self):
|
|
self._test_is_newer_than(timeutils.strtime)
|
|
|
|
def test_is_newer_than_aware(self):
|
|
"""Tests sending is_newer_than an 'aware' datetime."""
|
|
self._test_is_newer_than(lambda x: x.replace(
|
|
tzinfo=iso8601.iso8601.UTC))
|
|
|
|
def test_is_newer_than_aware_no_utc(self):
|
|
self._test_is_newer_than(lambda x: x.replace(
|
|
tzinfo=iso8601.iso8601.FixedOffset(1, 0, 'foo')).replace(
|
|
hour=7))
|
|
|
|
def test_set_time_override_using_default(self):
|
|
now = timeutils.utcnow_ts()
|
|
|
|
# NOTE(kgriffs): Normally it's bad form to sleep in a unit test,
|
|
# but this is the only way to test that set_time_override defaults
|
|
# to setting the override to the current time.
|
|
time.sleep(1)
|
|
|
|
timeutils.set_time_override()
|
|
overriden_now = timeutils.utcnow_ts()
|
|
self.assertThat(now, matchers.LessThan(overriden_now))
|
|
|
|
def test_utcnow_ts(self):
|
|
skynet_self_aware_ts = 872835240
|
|
skynet_dt = datetime.datetime.utcfromtimestamp(skynet_self_aware_ts)
|
|
self.assertEqual(self.skynet_self_aware_time, skynet_dt)
|
|
|
|
# NOTE(kgriffs): timeutils.utcnow_ts() uses time.time()
|
|
# IFF time override is not set.
|
|
with mock.patch('time.time') as time_mock:
|
|
time_mock.return_value = skynet_self_aware_ts
|
|
ts = timeutils.utcnow_ts()
|
|
self.assertEqual(ts, skynet_self_aware_ts)
|
|
|
|
timeutils.set_time_override(skynet_dt)
|
|
ts = timeutils.utcnow_ts()
|
|
self.assertEqual(ts, skynet_self_aware_ts)
|
|
|
|
def test_utcnow(self):
|
|
timeutils.set_time_override(mock.sentinel.utcnow)
|
|
self.assertEqual(timeutils.utcnow(), mock.sentinel.utcnow)
|
|
|
|
timeutils.clear_time_override()
|
|
self.assertFalse(timeutils.utcnow() == mock.sentinel.utcnow)
|
|
|
|
self.assertTrue(timeutils.utcnow())
|
|
|
|
def test_advance_time_delta(self):
|
|
timeutils.set_time_override(self.one_minute_before)
|
|
timeutils.advance_time_delta(datetime.timedelta(seconds=60))
|
|
self.assertEqual(timeutils.utcnow(), self.skynet_self_aware_time)
|
|
|
|
def test_advance_time_seconds(self):
|
|
timeutils.set_time_override(self.one_minute_before)
|
|
timeutils.advance_time_seconds(60)
|
|
self.assertEqual(timeutils.utcnow(), self.skynet_self_aware_time)
|
|
|
|
def test_marshall_time(self):
|
|
now = timeutils.utcnow()
|
|
binary = timeutils.marshall_now(now)
|
|
backagain = timeutils.unmarshall_time(binary)
|
|
self.assertEqual(now, backagain)
|
|
|
|
def test_marshall_time_with_tz(self):
|
|
now = timeutils.utcnow()
|
|
now = now.replace(tzinfo=iso8601.iso8601.UTC)
|
|
binary = timeutils.marshall_now(now)
|
|
self.assertEqual("UTC", binary['tzname'])
|
|
backagain = timeutils.unmarshall_time(binary)
|
|
self.assertEqual(now, backagain)
|
|
self.assertIsNotNone(backagain.tzinfo)
|
|
self.assertEqual(now.utcoffset(), backagain.utcoffset())
|
|
|
|
def test_unmarshall_time_leap_second(self):
|
|
leap_dict = dict(day=30, month=6, year=2015,
|
|
hour=23, minute=59,
|
|
second=timeutils._MAX_DATETIME_SEC + 1,
|
|
microsecond=0)
|
|
leap_time = timeutils.unmarshall_time(leap_dict)
|
|
|
|
leap_dict.update(second=timeutils._MAX_DATETIME_SEC)
|
|
expected = timeutils.unmarshall_time(leap_dict)
|
|
|
|
self.assertEqual(expected, leap_time)
|
|
|
|
def test_delta_seconds(self):
|
|
before = timeutils.utcnow()
|
|
after = before + datetime.timedelta(days=7, seconds=59,
|
|
microseconds=123456)
|
|
self.assertAlmostEquals(604859.123456,
|
|
timeutils.delta_seconds(before, after))
|
|
|
|
def test_iso8601_from_timestamp(self):
|
|
utcnow = timeutils.utcnow()
|
|
iso = timeutils.isotime(utcnow)
|
|
ts = calendar.timegm(utcnow.timetuple())
|
|
self.assertEqual(iso, timeutils.iso8601_from_timestamp(ts))
|
|
|
|
def test_iso8601_from_timestamp_ms(self):
|
|
ts = timeutils.utcnow_ts(microsecond=True)
|
|
utcnow = datetime.datetime.utcfromtimestamp(ts)
|
|
iso = timeutils.isotime(utcnow, subsecond=True)
|
|
self.assertEqual(iso, timeutils.iso8601_from_timestamp(ts, True))
|
|
|
|
def test_is_soon(self):
|
|
expires = timeutils.utcnow() + datetime.timedelta(minutes=5)
|
|
self.assertFalse(timeutils.is_soon(expires, 120))
|
|
self.assertTrue(timeutils.is_soon(expires, 300))
|
|
self.assertTrue(timeutils.is_soon(expires, 600))
|
|
|
|
with mock.patch('datetime.datetime') as datetime_mock:
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_time
|
|
expires = timeutils.utcnow()
|
|
self.assertTrue(timeutils.is_soon(expires, 0))
|
|
|
|
|
|
class TestIso8601Time(test_base.BaseTestCase):
|
|
|
|
def _instaneous(self, timestamp, yr, mon, day, hr, minute, sec, micro):
|
|
self.assertEqual(timestamp.year, yr)
|
|
self.assertEqual(timestamp.month, mon)
|
|
self.assertEqual(timestamp.day, day)
|
|
self.assertEqual(timestamp.hour, hr)
|
|
self.assertEqual(timestamp.minute, minute)
|
|
self.assertEqual(timestamp.second, sec)
|
|
self.assertEqual(timestamp.microsecond, micro)
|
|
|
|
def _do_test(self, time_str, yr, mon, day, hr, minute, sec, micro, shift):
|
|
DAY_SECONDS = 24 * 60 * 60
|
|
timestamp = timeutils.parse_isotime(time_str)
|
|
self._instaneous(timestamp, yr, mon, day, hr, minute, sec, micro)
|
|
offset = timestamp.tzinfo.utcoffset(None)
|
|
self.assertEqual(offset.seconds + offset.days * DAY_SECONDS, shift)
|
|
|
|
def test_zulu(self):
|
|
time_str = '2012-02-14T20:53:07Z'
|
|
self._do_test(time_str, 2012, 2, 14, 20, 53, 7, 0, 0)
|
|
|
|
def test_zulu_micros(self):
|
|
time_str = '2012-02-14T20:53:07.123Z'
|
|
self._do_test(time_str, 2012, 2, 14, 20, 53, 7, 123000, 0)
|
|
|
|
def test_offset_east(self):
|
|
time_str = '2012-02-14T20:53:07+04:30'
|
|
offset = 4.5 * 60 * 60
|
|
self._do_test(time_str, 2012, 2, 14, 20, 53, 7, 0, offset)
|
|
|
|
def test_offset_east_micros(self):
|
|
time_str = '2012-02-14T20:53:07.42+04:30'
|
|
offset = 4.5 * 60 * 60
|
|
self._do_test(time_str, 2012, 2, 14, 20, 53, 7, 420000, offset)
|
|
|
|
def test_offset_west(self):
|
|
time_str = '2012-02-14T20:53:07-05:30'
|
|
offset = -5.5 * 60 * 60
|
|
self._do_test(time_str, 2012, 2, 14, 20, 53, 7, 0, offset)
|
|
|
|
def test_offset_west_micros(self):
|
|
time_str = '2012-02-14T20:53:07.654321-05:30'
|
|
offset = -5.5 * 60 * 60
|
|
self._do_test(time_str, 2012, 2, 14, 20, 53, 7, 654321, offset)
|
|
|
|
def test_compare(self):
|
|
zulu = timeutils.parse_isotime('2012-02-14T20:53:07')
|
|
east = timeutils.parse_isotime('2012-02-14T20:53:07-01:00')
|
|
west = timeutils.parse_isotime('2012-02-14T20:53:07+01:00')
|
|
self.assertTrue(east > west)
|
|
self.assertTrue(east > zulu)
|
|
self.assertTrue(zulu > west)
|
|
|
|
def test_compare_micros(self):
|
|
zulu = timeutils.parse_isotime('2012-02-14T20:53:07.6544')
|
|
east = timeutils.parse_isotime('2012-02-14T19:53:07.654321-01:00')
|
|
west = timeutils.parse_isotime('2012-02-14T21:53:07.655+01:00')
|
|
self.assertTrue(east < west)
|
|
self.assertTrue(east < zulu)
|
|
self.assertTrue(zulu < west)
|
|
|
|
def test_zulu_roundtrip(self):
|
|
time_str = '2012-02-14T20:53:07Z'
|
|
zulu = timeutils.parse_isotime(time_str)
|
|
self.assertEqual(zulu.tzinfo, iso8601.iso8601.UTC)
|
|
self.assertEqual(timeutils.isotime(zulu), time_str)
|
|
|
|
def test_east_roundtrip(self):
|
|
time_str = '2012-02-14T20:53:07-07:00'
|
|
east = timeutils.parse_isotime(time_str)
|
|
self.assertEqual(east.tzinfo.tzname(None), '-07:00')
|
|
self.assertEqual(timeutils.isotime(east), time_str)
|
|
|
|
def test_west_roundtrip(self):
|
|
time_str = '2012-02-14T20:53:07+11:30'
|
|
west = timeutils.parse_isotime(time_str)
|
|
self.assertEqual(west.tzinfo.tzname(None), '+11:30')
|
|
self.assertEqual(timeutils.isotime(west), time_str)
|
|
|
|
def test_now_roundtrip(self):
|
|
time_str = timeutils.isotime()
|
|
now = timeutils.parse_isotime(time_str)
|
|
self.assertEqual(now.tzinfo, iso8601.iso8601.UTC)
|
|
self.assertEqual(timeutils.isotime(now), time_str)
|
|
|
|
def test_zulu_normalize(self):
|
|
time_str = '2012-02-14T20:53:07Z'
|
|
zulu = timeutils.parse_isotime(time_str)
|
|
normed = timeutils.normalize_time(zulu)
|
|
self._instaneous(normed, 2012, 2, 14, 20, 53, 7, 0)
|
|
|
|
def test_east_normalize(self):
|
|
time_str = '2012-02-14T20:53:07-07:00'
|
|
east = timeutils.parse_isotime(time_str)
|
|
normed = timeutils.normalize_time(east)
|
|
self._instaneous(normed, 2012, 2, 15, 3, 53, 7, 0)
|
|
|
|
def test_west_normalize(self):
|
|
time_str = '2012-02-14T20:53:07+21:00'
|
|
west = timeutils.parse_isotime(time_str)
|
|
normed = timeutils.normalize_time(west)
|
|
self._instaneous(normed, 2012, 2, 13, 23, 53, 7, 0)
|
|
|
|
def test_normalize_aware_to_naive(self):
|
|
dt = datetime.datetime(2011, 2, 14, 20, 53, 7)
|
|
time_str = '2011-02-14T20:53:07+21:00'
|
|
aware = timeutils.parse_isotime(time_str)
|
|
naive = timeutils.normalize_time(aware)
|
|
self.assertTrue(naive < dt)
|
|
|
|
def test_normalize_zulu_aware_to_naive(self):
|
|
dt = datetime.datetime(2011, 2, 14, 20, 53, 7)
|
|
time_str = '2011-02-14T19:53:07Z'
|
|
aware = timeutils.parse_isotime(time_str)
|
|
naive = timeutils.normalize_time(aware)
|
|
self.assertTrue(naive < dt)
|
|
|
|
def test_normalize_naive(self):
|
|
dt = datetime.datetime(2011, 2, 14, 20, 53, 7)
|
|
dtn = datetime.datetime(2011, 2, 14, 19, 53, 7)
|
|
naive = timeutils.normalize_time(dtn)
|
|
self.assertTrue(naive < dt)
|
|
|
|
|
|
class TimeItTest(test_base.BaseTestCase):
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_timed(self, mock_now, mock_sleep):
|
|
mock_now.side_effect = monotonic_iter(incr=0.1)
|
|
fake_logger = mock.MagicMock(logging.getLogger(), autospec=True)
|
|
|
|
@timeutils.time_it(fake_logger)
|
|
def slow_function():
|
|
time.sleep(0.1)
|
|
|
|
slow_function()
|
|
self.assertTrue(mock_now.called)
|
|
self.assertTrue(mock_sleep.called)
|
|
self.assertTrue(fake_logger.log.called)
|
|
fake_logger.log.assert_called_with(logging.DEBUG, mock.ANY, mock.ANY)
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_no_timed_disabled(self, mock_now, mock_sleep):
|
|
mock_now.side_effect = monotonic_iter(incr=0.1)
|
|
fake_logger = mock.MagicMock(logging.getLogger(), autospec=True)
|
|
|
|
@timeutils.time_it(fake_logger, enabled=False)
|
|
def slow_function():
|
|
time.sleep(0.1)
|
|
|
|
slow_function()
|
|
self.assertFalse(mock_now.called)
|
|
self.assertFalse(fake_logger.log.called)
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_no_timed_to_fast(self, mock_now, mock_sleep):
|
|
mock_now.side_effect = monotonic_iter(incr=0.1)
|
|
fake_logger = mock.MagicMock(logging.getLogger(), autospec=True)
|
|
|
|
@timeutils.time_it(fake_logger, min_duration=10)
|
|
def fast_function():
|
|
pass
|
|
|
|
fast_function()
|
|
self.assertFalse(fake_logger.log.called)
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_no_timed_exception(self, mock_now, mock_sleep):
|
|
mock_now.side_effect = monotonic_iter(incr=0.1)
|
|
fake_logger = mock.MagicMock(logging.getLogger(), autospec=True)
|
|
|
|
@timeutils.time_it(fake_logger)
|
|
def broken_function():
|
|
raise IOError("Broken")
|
|
|
|
self.assertRaises(IOError, broken_function)
|
|
self.assertFalse(fake_logger.log.called)
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_timed_custom_message(self, mock_now, mock_sleep):
|
|
mock_now.side_effect = monotonic_iter(incr=0.1)
|
|
fake_logger = mock.MagicMock(logging.getLogger(), autospec=True)
|
|
|
|
@timeutils.time_it(fake_logger, message="That took a long time")
|
|
def slow_function():
|
|
time.sleep(0.1)
|
|
|
|
slow_function()
|
|
self.assertTrue(mock_now.called)
|
|
self.assertTrue(mock_sleep.called)
|
|
self.assertTrue(fake_logger.log.called)
|
|
fake_logger.log.assert_called_with(logging.DEBUG,
|
|
"That took a long time",
|
|
mock.ANY)
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_timed_custom_level(self, mock_now, mock_sleep):
|
|
mock_now.side_effect = monotonic_iter(incr=0.1)
|
|
fake_logger = mock.MagicMock(logging.getLogger(), autospec=True)
|
|
|
|
@timeutils.time_it(fake_logger, log_level=logging.INFO)
|
|
def slow_function():
|
|
time.sleep(0.1)
|
|
|
|
slow_function()
|
|
self.assertTrue(mock_now.called)
|
|
self.assertTrue(mock_sleep.called)
|
|
self.assertTrue(fake_logger.log.called)
|
|
fake_logger.log.assert_called_with(logging.INFO, mock.ANY, mock.ANY)
|
|
|
|
|
|
class StopWatchTest(test_base.BaseTestCase):
|
|
def test_leftover_no_duration(self):
|
|
watch = timeutils.StopWatch()
|
|
watch.start()
|
|
self.assertRaises(RuntimeError, watch.leftover)
|
|
self.assertRaises(RuntimeError, watch.leftover, return_none=False)
|
|
self.assertIsNone(watch.leftover(return_none=True))
|
|
|
|
def test_no_states(self):
|
|
watch = timeutils.StopWatch()
|
|
self.assertRaises(RuntimeError, watch.stop)
|
|
self.assertRaises(RuntimeError, watch.resume)
|
|
|
|
def test_bad_expiry(self):
|
|
self.assertRaises(ValueError, timeutils.StopWatch, -1)
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_backwards(self, mock_now):
|
|
mock_now.side_effect = [0, 0.5, -1.0, -1.0]
|
|
watch = timeutils.StopWatch(0.1)
|
|
watch.start()
|
|
self.assertTrue(watch.expired())
|
|
self.assertFalse(watch.expired())
|
|
self.assertEqual(0.0, watch.elapsed())
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_expiry(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter(incr=0.2)
|
|
watch = timeutils.StopWatch(0.1)
|
|
watch.start()
|
|
self.assertTrue(watch.expired())
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_not_expired(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter()
|
|
watch = timeutils.StopWatch(0.1)
|
|
watch.start()
|
|
self.assertFalse(watch.expired())
|
|
|
|
def test_has_started_stopped(self):
|
|
watch = timeutils.StopWatch()
|
|
self.assertFalse(watch.has_started())
|
|
self.assertFalse(watch.has_stopped())
|
|
watch.start()
|
|
|
|
self.assertTrue(watch.has_started())
|
|
self.assertFalse(watch.has_stopped())
|
|
|
|
watch.stop()
|
|
self.assertTrue(watch.has_stopped())
|
|
self.assertFalse(watch.has_started())
|
|
|
|
def test_no_expiry(self):
|
|
watch = timeutils.StopWatch(0.1)
|
|
self.assertRaises(RuntimeError, watch.expired)
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_elapsed(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter(incr=0.2)
|
|
watch = timeutils.StopWatch()
|
|
watch.start()
|
|
matcher = matchers.GreaterThan(0.19)
|
|
self.assertThat(watch.elapsed(), matcher)
|
|
|
|
def test_no_elapsed(self):
|
|
watch = timeutils.StopWatch()
|
|
self.assertRaises(RuntimeError, watch.elapsed)
|
|
|
|
def test_no_leftover(self):
|
|
watch = timeutils.StopWatch()
|
|
self.assertRaises(RuntimeError, watch.leftover)
|
|
watch = timeutils.StopWatch(1)
|
|
self.assertRaises(RuntimeError, watch.leftover)
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_pause_resume(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter()
|
|
watch = timeutils.StopWatch()
|
|
watch.start()
|
|
watch.stop()
|
|
elapsed = watch.elapsed()
|
|
self.assertAlmostEqual(elapsed, watch.elapsed())
|
|
watch.resume()
|
|
self.assertNotEqual(elapsed, watch.elapsed())
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_context_manager(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter()
|
|
with timeutils.StopWatch() as watch:
|
|
pass
|
|
matcher = matchers.GreaterThan(0.04)
|
|
self.assertThat(watch.elapsed(), matcher)
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_context_manager_splits(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter()
|
|
with timeutils.StopWatch() as watch:
|
|
time.sleep(0.01)
|
|
watch.split()
|
|
self.assertRaises(RuntimeError, watch.split)
|
|
self.assertEqual(1, len(watch.splits))
|
|
|
|
def test_splits_stopped(self):
|
|
watch = timeutils.StopWatch()
|
|
watch.start()
|
|
watch.split()
|
|
watch.stop()
|
|
self.assertRaises(RuntimeError, watch.split)
|
|
|
|
def test_splits_never_started(self):
|
|
watch = timeutils.StopWatch()
|
|
self.assertRaises(RuntimeError, watch.split)
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_splits(self, mock_now):
|
|
mock_now.side_effect = monotonic_iter()
|
|
|
|
watch = timeutils.StopWatch()
|
|
watch.start()
|
|
self.assertEqual(0, len(watch.splits))
|
|
|
|
watch.split()
|
|
self.assertEqual(1, len(watch.splits))
|
|
self.assertEqual(watch.splits[0].elapsed,
|
|
watch.splits[0].length)
|
|
|
|
watch.split()
|
|
splits = watch.splits
|
|
self.assertEqual(2, len(splits))
|
|
self.assertNotEqual(splits[0].elapsed, splits[1].elapsed)
|
|
self.assertEqual(splits[1].length,
|
|
splits[1].elapsed - splits[0].elapsed)
|
|
|
|
watch.stop()
|
|
self.assertEqual(2, len(watch.splits))
|
|
|
|
watch.start()
|
|
self.assertEqual(0, len(watch.splits))
|
|
|
|
@mock.patch('oslo_utils.timeutils.now')
|
|
def test_elapsed_maximum(self, mock_now):
|
|
mock_now.side_effect = [0, 1] + ([11] * 4)
|
|
|
|
watch = timeutils.StopWatch()
|
|
watch.start()
|
|
self.assertEqual(1, watch.elapsed())
|
|
|
|
self.assertEqual(11, watch.elapsed())
|
|
self.assertEqual(1, watch.elapsed(maximum=1))
|
|
|
|
watch.stop()
|
|
self.assertEqual(11, watch.elapsed())
|
|
self.assertEqual(11, watch.elapsed())
|
|
self.assertEqual(0, watch.elapsed(maximum=-1))
|