diff options
Diffstat (limited to 'Lib')
| -rw-r--r-- | Lib/queue.py | 5 | ||||
| -rw-r--r-- | Lib/test/test_time.py | 107 | ||||
| -rw-r--r-- | Lib/threading.py | 6 |
3 files changed, 104 insertions, 14 deletions
diff --git a/Lib/queue.py b/Lib/queue.py index 1dc72c4..c3296fe 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -6,7 +6,10 @@ except ImportError: import dummy_threading as threading from collections import deque from heapq import heappush, heappop -from time import steady as time +try: + from time import monotonic as time +except ImportError: + from time import time __all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue'] diff --git a/Lib/test/test_time.py b/Lib/test/test_time.py index fb2489c..f00660b 100644 --- a/Lib/test/test_time.py +++ b/Lib/test/test_time.py @@ -5,6 +5,10 @@ import locale import sysconfig import sys import platform +try: + import threading +except ImportError: + threading = None # Max year is only limited by the size of C int. SIZEOF_INT = sysconfig.get_config_var('SIZEOF_INT') or 4 @@ -23,9 +27,20 @@ class TimeTestCase(unittest.TestCase): time.timezone time.tzname + def test_time(self): + time.time() + info = time.get_clock_info('time') + self.assertEqual(info.is_monotonic, False) + if sys.platform != 'win32': + self.assertEqual(info.is_adjusted, True) + def test_clock(self): time.clock() + info = time.get_clock_info('clock') + self.assertEqual(info.is_monotonic, True) + self.assertEqual(info.is_adjusted, False) + @unittest.skipUnless(hasattr(time, 'clock_gettime'), 'need time.clock_gettime()') def test_clock_realtime(self): @@ -56,7 +71,9 @@ class TimeTestCase(unittest.TestCase): except PermissionError: pass - self.assertRaises(OSError, time.clock_settime, time.CLOCK_MONOTONIC, 0) + if hasattr(time, 'CLOCK_MONOTONIC'): + self.assertRaises(OSError, + time.clock_settime, time.CLOCK_MONOTONIC, 0) def test_conversions(self): self.assertEqual(time.ctime(self.t), @@ -342,23 +359,69 @@ class TimeTestCase(unittest.TestCase): pass self.assertEqual(time.strftime('%Z', tt), tzname) - def test_steady(self): - t1 = time.steady() + @unittest.skipUnless(hasattr(time, 'monotonic'), + 'need time.monotonic') + def test_monotonic(self): + t1 = time.monotonic() time.sleep(0.1) - t2 = time.steady() + t2 = time.monotonic() dt = t2 - t1 - # may fail if the system clock was changed self.assertGreater(t2, t1) self.assertAlmostEqual(dt, 0.1, delta=0.2) - def test_steady_strict(self): + info = time.get_clock_info('monotonic') + self.assertEqual(info.is_monotonic, True) + if sys.platform == 'linux': + self.assertEqual(info.is_adjusted, True) + else: + self.assertEqual(info.is_adjusted, False) + + def test_perf_counter(self): + time.perf_counter() + + def test_process_time(self): + start = time.process_time() + time.sleep(0.1) + stop = time.process_time() + self.assertLess(stop - start, 0.01) + + info = time.get_clock_info('process_time') + self.assertEqual(info.is_monotonic, True) + self.assertEqual(info.is_adjusted, False) + + @unittest.skipUnless(threading, + 'need threading') + def test_process_time_threads(self): + class BusyThread(threading.Thread): + def run(self): + while not self.stop: + pass + + thread = BusyThread() + thread.stop = False + t1 = time.process_time() + thread.start() + time.sleep(0.2) + t2 = time.process_time() + thread.stop = True + thread.join() + self.assertGreater(t2 - t1, 0.1) + + @unittest.skipUnless(hasattr(time, 'monotonic'), + 'need time.monotonic') + @unittest.skipUnless(hasattr(time, 'clock_settime'), + 'need time.clock_settime') + def test_monotonic_settime(self): + t1 = time.monotonic() + realtime = time.clock_gettime(time.CLOCK_REALTIME) + # jump backward with an offset of 1 hour try: - t1 = time.steady(strict=True) - except OSError as err: - self.skipTest("the monotonic clock failed: %s" % err) - except NotImplementedError: - self.skipTest("no monotonic clock available") - t2 = time.steady(strict=True) + time.clock_settime(time.CLOCK_REALTIME, realtime - 3600) + except PermissionError as err: + self.skipTest(err) + t2 = time.monotonic() + time.clock_settime(time.CLOCK_REALTIME, realtime) + # monotonic must not be affected by system clock updates self.assertGreaterEqual(t2, t1) def test_localtime_failure(self): @@ -378,6 +441,26 @@ class TimeTestCase(unittest.TestCase): self.assertRaises(OSError, time.localtime, invalid_time_t) self.assertRaises(OSError, time.ctime, invalid_time_t) + def test_get_clock_info(self): + clocks = ['clock', 'perf_counter', 'process_time', 'time'] + if hasattr(time, 'monotonic'): + clocks.append('monotonic') + + for name in clocks: + info = time.get_clock_info(name) + #self.assertIsInstance(info, dict) + self.assertIsInstance(info.implementation, str) + self.assertNotEqual(info.implementation, '') + self.assertIsInstance(info.is_monotonic, bool) + self.assertIsInstance(info.resolution, float) + # 0.0 < resolution <= 1.0 + self.assertGreater(info.resolution, 0.0) + self.assertLessEqual(info.resolution, 1.0) + self.assertIsInstance(info.is_adjusted, bool) + + self.assertRaises(ValueError, time.get_clock_info, 'xxx') + + class TestLocale(unittest.TestCase): def setUp(self): self.oldloc = locale.setlocale(locale.LC_ALL) diff --git a/Lib/threading.py b/Lib/threading.py index 8c2cee9..6c34d49 100644 --- a/Lib/threading.py +++ b/Lib/threading.py @@ -3,7 +3,11 @@ import sys as _sys import _thread -from time import steady as _time, sleep as _sleep +from time import sleep as _sleep +try: + from time import monotonic as _time +except ImportError: + from time import time as _time from traceback import format_exc as _format_exc from _weakrefset import WeakSet |