5a8cade630
I have improved the readability of the code by fixing the
problems that flake8 reported.
The type of errors that are fixed are:
* H302 - Import only module
* H305 - Import not grouped correctly
* H307 - Like imports shoud be grouped together
* H402 - One line docstring needs punctation
* F821 - Undefined name
Change-Id: I3a7f12120b6a99342a3d2025206edcdc0f6a4a93
Signed-off-by: Costin Galan <cgalan@cloudbasesolutions.com>
630 lines
24 KiB
Python
630 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)
|
|
|
|
def _test_is_older_than(self, fn):
|
|
strptime = datetime.datetime.strptime
|
|
with mock.patch('datetime.datetime') as datetime_mock:
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_time
|
|
datetime_mock.strptime = strptime
|
|
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))
|
|
|
|
def _test_is_newer_than(self, fn):
|
|
strptime = datetime.datetime.strptime
|
|
with mock.patch('datetime.datetime') as datetime_mock:
|
|
datetime_mock.utcnow.return_value = self.skynet_self_aware_time
|
|
datetime_mock.strptime = strptime
|
|
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))
|