From 659e12bddd638500a17e9b38effbaded134626e3 Mon Sep 17 00:00:00 2001 From: Paul Belanger Date: Thu, 15 Jan 2015 22:21:14 -0500 Subject: [PATCH] Add microsecond support to timeutils.utcnow_ts() Now we have the ability to generate a timestamp with microseconds. Change-Id: I34d51b71c01df02bef46639149fde08f89f0e110 Signed-off-by: Paul Belanger --- oslo_utils/timeutils.py | 15 ++++++++++++--- tests/test_timeutils.py | 16 ++++++++++++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/oslo_utils/timeutils.py b/oslo_utils/timeutils.py index c48da95f..fac739b2 100644 --- a/oslo_utils/timeutils.py +++ b/oslo_utils/timeutils.py @@ -93,14 +93,23 @@ def is_newer_than(after, seconds): return after - utcnow() > datetime.timedelta(seconds=seconds) -def utcnow_ts(): +def utcnow_ts(microsecond=False): """Timestamp version of our utcnow function.""" if utcnow.override_time is None: # NOTE(kgriffs): This is several times faster # than going through calendar.timegm(...) - return int(time.time()) + timestamp = time.time() + if not microsecond: + timestamp = int(timestamp) + return timestamp - return calendar.timegm(utcnow().timetuple()) + now = utcnow() + timestamp = calendar.timegm(now.timetuple()) + + if microsecond: + timestamp += float(now.microsecond) / 1000000 + + return timestamp def utcnow(): diff --git a/tests/test_timeutils.py b/tests/test_timeutils.py index fef980f5..fdfab2cc 100644 --- a/tests/test_timeutils.py +++ b/tests/test_timeutils.py @@ -160,6 +160,22 @@ class TimeUtilsTest(test_base.BaseTestCase): ts = timeutils.utcnow_ts() self.assertEqual(ts, skynet_self_aware_ts) + def test_utcnow_ts_microsecond(self): + skynet_self_aware_ts = 872835240.000123 + skynet_dt = datetime.datetime.utcfromtimestamp(skynet_self_aware_ts) + self.assertEqual(self.skynet_self_aware_ms_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(microsecond=True) + self.assertEqual(ts, skynet_self_aware_ts) + + timeutils.set_time_override(skynet_dt) + ts = timeutils.utcnow_ts(microsecond=True) + 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)