Repository: pika/pika

Issue:
Use monotonic time where appropriate
See @vitaly-krugl 's comment [here](https://github.com/pika/pika/pull/1045#issuecomment-389207040)

Discussion:

Patch:
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index db02330..f185328 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -789,7 +789,7 @@ class BlockingConnection(object):
         LOGGER.info('Closing connection (%s): %s', reply_code, reply_text)
 
         # Close channels that remain opened
-        for impl_channel in pika.compat.dictvalues(self._impl._channels):
+        for impl_channel in compat.dictvalues(self._impl._channels):
             channel = impl_channel._get_cookie()
             if channel.is_open:
                 try:
@@ -845,12 +845,12 @@ class BlockingConnection(object):
         """
         assert duration >= 0, duration
 
-        deadline = time.time() + duration
+        deadline = compat.time_now() + duration
         time_limit = duration
         # Process events at least once
         while True:
             self.process_data_events(time_limit)
-            time_limit = deadline - time.time()
+            time_limit = deadline - compat.time_now()
             if time_limit <= 0:
                 break
 
@@ -1467,7 +1467,7 @@ class BlockingChannel(object):
                 self.cancel()
 
             # Cancel consumers created via basic_consume
-            for consumer_tag in pika.compat.dictkeys(self._consumer_infos):
+            for consumer_tag in compat.dictkeys(self._consumer_infos):
                 self.basic_cancel(consumer_tag)
 
     def _dispatch_events(self):
@@ -1973,7 +1973,7 @@ class BlockingChannel(object):
                 continue
 
             # Wait with inactivity timeout
-            wait_start_time = time.time()
+            wait_start_time = compat.time_now()
             wait_deadline = wait_start_time + inactivity_timeout
             delta = inactivity_timeout
 
@@ -1990,7 +1990,7 @@ class BlockingChannel(object):
                     # Got message(s)
                     break
 
-                delta = wait_deadline - time.time()
+                delta = wait_deadline - compat.time_now()
                 if delta <= 0.0:
                     # Signal inactivity timeout
                     yield (None, None, None)
diff --git a/pika/adapters/select_connection.py b/pika/adapters/select_connection.py
index f1c1617..4e05af3 100644
--- a/pika/adapters/select_connection.py
+++ b/pika/adapters/select_connection.py
@@ -251,7 +251,7 @@ class _Timer(object):
             raise ValueError(
                 'call_later: delay must be non-negative, but got %r' % (delay,))
 
-        now = time.time()
+        now = pika.compat.time_now()
 
         timeout = _Timeout(now + delay, callback)
 
@@ -293,7 +293,8 @@ class _Timer(object):
 
         """
         if self._timeout_heap:
-            interval = max(0, self._timeout_heap[0].deadline - time.time())
+            now = pika.compat.time_now()
+            interval = max(0, self._timeout_heap[0].deadline - now)
         else:
             interval = None
 
@@ -305,7 +306,7 @@ class _Timer(object):
 
         """
         if self._timeout_heap:
-            now = time.time()
+            now = pika.compat.time_now()
 
             # Remove ready timeouts from the heap now to prevent IO starvation
             # from timeouts added during callback processing
diff --git a/pika/compat.py b/pika/compat.py
index c1a4ac7..c0633c1 100644
--- a/pika/compat.py
+++ b/pika/compat.py
@@ -10,6 +10,7 @@ import platform
 import re
 import socket
 import sys as _sys
+import time
 
 PY2 = _sys.version_info.major == 2
 PY3 = not PY2
@@ -53,6 +54,12 @@ if PY3:
     # the unicode type is str
     unicode_type = str
 
+    def time_now():
+        """
+        Python 3 supports monotonic time
+        """
+        return time.monotonic()
+
     def dictkeys(dct):
         """
         Returns a list of keys of dictionary
@@ -143,6 +150,12 @@ else:
     byte = chr
     long = long
 
+    def time_now():
+        """
+        Python 2 does not support monotonic time
+        """
+        return time.time()
+
     def canonical_str(value):
         """
         Returns the canonical string value of the given string.


Test patch:
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index 014aa4b..a2ce8df 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -27,7 +27,7 @@ import uuid
 import pika
 from pika.adapters.utils import connection_workflow
 from pika import spec
-from pika.compat import as_bytes
+from pika.compat import as_bytes, time_now
 import pika.connection
 import pika.exceptions
 import pika.frame
@@ -550,7 +550,7 @@ class TestConsumeCancel(AsyncTestCase, AsyncAdapters):
     def on_queue_declared(self, frame):
         for i in range(0, 100):
             msg_body = '{}:{}:{}'.format(self.__class__.__name__, i,
-                                         time.time())
+                                         time_now())
             self.channel.basic_publish('', self.queue_name, msg_body)
         self.ctag = self.channel.basic_consume(self.queue_name,
                                                self.on_message,
@@ -854,7 +854,7 @@ class TestZ_PublishAndConsume(BoundQueueTestCase, AsyncAdapters):  # pylint: dis
 
     def on_ready(self, frame):
         self.ctag = self.channel.basic_consume(self.queue, self.on_message)
-        self.msg_body = "%s: %i" % (self.__class__.__name__, time.time())
+        self.msg_body = "%s: %i" % (self.__class__.__name__, time_now())
         self.channel.basic_publish(self.exchange, self.routing_key,
                                    self.msg_body)
 
@@ -897,7 +897,7 @@ class TestZ_PublishAndGet(BoundQueueTestCase, AsyncAdapters):  # pylint: disable
     DESCRIPTION = "Publish a message and get it"
 
     def on_ready(self, frame):
-        self.msg_body = "%s: %i" % (self.__class__.__name__, time.time())
+        self.msg_body = "%s: %i" % (self.__class__.__name__, time_now())
         self.channel.basic_publish(self.exchange, self.routing_key,
                                    self.msg_body)
         self.channel.basic_get(self.queue, self.on_get)
@@ -1002,7 +1002,7 @@ class TestAddCallbackThreadsafeRequestBeforeIOLoopStarts(AsyncTestCase, AsyncAda
         _adapter_add_callback_threadsafe before AsyncTestCase starts the ioloop.
 
         """
-        self.my_start_time = time.time()
+        self.my_start_time = time_now()
         # Request a callback from our current (ioloop's) thread
         self.connection._adapter_add_callback_threadsafe(
             self.on_requested_callback)
@@ -1024,7 +1024,7 @@ class TestAddCallbackThreadsafeRequestBeforeIOLoopStarts(AsyncTestCase, AsyncAda
     def on_requested_callback(self):
         self.assertEqual(threading.current_thread().ident,
                          self.loop_thread_ident)
-        self.assertLess(time.time() - self.my_start_time, 0.25)
+        self.assertLess(time_now() - self.my_start_time, 0.25)
         self.got_callback = True
 
 
@@ -1040,7 +1040,7 @@ class TestAddCallbackThreadsafeFromIOLoopThread(AsyncTestCase, AsyncAdapters):
         self.assertTrue(self.got_callback)
 
     def begin(self, channel):
-        self.my_start_time = time.time()
+        self.my_start_time = time_now()
         # Request a callback from our current (ioloop's) thread
         channel.connection._adapter_add_callback_threadsafe(
             self.on_requested_callback)
@@ -1048,7 +1048,7 @@ class TestAddCallbackThreadsafeFromIOLoopThread(AsyncTestCase, AsyncAdapters):
     def on_requested_callback(self):
         self.assertEqual(threading.current_thread().ident,
                          self.loop_thread_ident)
-        self.assertLess(time.time() - self.my_start_time, 0.25)
+        self.assertLess(time_now() - self.my_start_time, 0.25)
         self.got_callback = True
         self.stop()
 
@@ -1065,7 +1065,7 @@ class TestAddCallbackThreadsafeFromAnotherThread(AsyncTestCase, AsyncAdapters):
         self.assertTrue(self.got_callback)
 
     def begin(self, channel):
-        self.my_start_time = time.time()
+        self.my_start_time = time_now()
         # Request a callback from ioloop while executing in another thread
         timer = threading.Timer(
             0,
@@ -1077,7 +1077,7 @@ class TestAddCallbackThreadsafeFromAnotherThread(AsyncTestCase, AsyncAdapters):
     def on_requested_callback(self):
         self.assertEqual(threading.current_thread().ident,
                          self.loop_thread_ident)
-        self.assertLess(time.time() - self.my_start_time, 0.25)
+        self.assertLess(time_now() - self.my_start_time, 0.25)
         self.got_callback = True
         self.stop()
 
@@ -1090,13 +1090,13 @@ class TestIOLoopStopBeforeIOLoopStarts(AsyncTestCase, AsyncAdapters):
         ioloop.stop() before AsyncTestCase starts the ioloop.
         """
         # Request ioloop to stop before it starts
-        my_start_time = time.time()
+        my_start_time = time_now()
         self.stop_ioloop_only()
 
         super(
             TestIOLoopStopBeforeIOLoopStarts, self)._run_ioloop(*args, **kwargs)
 
-        self.assertLess(time.time() - my_start_time, 0.25)
+        self.assertLess(time_now() - my_start_time, 0.25)
 
         # Resume I/O loop to facilitate normal course of events and closure
         # of connection in order to prevent reporting of a socket resource leak
diff --git a/tests/acceptance/blocking_adapter_test.py b/tests/acceptance/blocking_adapter_test.py
index 22b63d2..2b2ee2a 100644
--- a/tests/acceptance/blocking_adapter_test.py
+++ b/tests/acceptance/blocking_adapter_test.py
@@ -10,7 +10,7 @@ import uuid
 
 import pika
 from pika.adapters import blocking_connection
-from pika.compat import as_bytes
+from pika.compat import as_bytes, time_now
 import pika.connection
 import pika.exceptions
 
@@ -519,15 +519,15 @@ class TestProcessDataEvents(BlockingTestCaseBase):
         connection = self._connect()
 
         # Try with time_limit=0
-        start_time = time.time()
+        start_time = time_now()
         connection.process_data_events(time_limit=0)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertLess(elapsed, 0.25)
 
         # Try with time_limit=0.005
-        start_time = time.time()
+        start_time = time_now()
         connection.process_data_events(time_limit=0.005)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertGreaterEqual(elapsed, 0.005)
         self.assertLess(elapsed, 0.25)
 
@@ -601,15 +601,15 @@ class TestAddCallbackThreadsafeFromSameThread(BlockingTestCaseBase):
         connection = self._connect()
 
         # Test timer completion
-        start_time = time.time()
+        start_time = time_now()
         rx_callback = []
         connection.add_callback_threadsafe(
-            lambda: rx_callback.append(time.time()))
+            lambda: rx_callback.append(time_now()))
         while not rx_callback:
             connection.process_data_events(time_limit=None)
 
         self.assertEqual(len(rx_callback), 1)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertLess(elapsed, 0.25)
 
 
@@ -620,19 +620,19 @@ class TestAddCallbackThreadsafeFromAnotherThread(BlockingTestCaseBase):
         connection = self._connect()
 
         # Test timer completion
-        start_time = time.time()
+        start_time = time_now()
         rx_callback = []
         timer = threading.Timer(
             0,
             functools.partial(connection.add_callback_threadsafe,
-                              lambda: rx_callback.append(time.time())))
+                              lambda: rx_callback.append(time_now())))
         self.addCleanup(timer.cancel)
         timer.start()
         while not rx_callback:
             connection.process_data_events(time_limit=None)
 
         self.assertEqual(len(rx_callback), 1)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertLess(elapsed, 0.25)
 
 
@@ -655,16 +655,16 @@ class TestAddTimeoutRemoveTimeout(BlockingTestCaseBase):
         connection = self._connect()
 
         # Test timer completion
-        start_time = time.time()
+        start_time = time_now()
         rx_callback = []
         timer_id = connection.call_later(
             0.005,
-            lambda: rx_callback.append(time.time()))
+            lambda: rx_callback.append(time_now()))
         while not rx_callback:
             connection.process_data_events(time_limit=None)
 
         self.assertEqual(len(rx_callback), 1)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertLess(elapsed, 0.25)
 
         # Test removing triggered timeout
@@ -675,7 +675,7 @@ class TestAddTimeoutRemoveTimeout(BlockingTestCaseBase):
         rx_callback = []
         timer_id = connection.call_later(
             0.001,
-            lambda: rx_callback.append(time.time()))
+            lambda: rx_callback.append(time_now()))
         connection.remove_timeout(timer_id)
         connection.process_data_events(time_limit=0.1)
         self.assertFalse(rx_callback)
@@ -704,8 +704,8 @@ class TestViabilityOfMultipleTimeoutsWithSameDeadlineAndCallback(BlockingTestCas
         connection.remove_timeout(timer1)
 
         # Wait for second timer to fire
-        start_wait_time = time.time()
-        while not rx_callback and time.time() - start_wait_time < 0.25:
+        start_wait_time = time_now()
+        while not rx_callback and time_now() - start_wait_time < 0.25:
             connection.process_data_events(time_limit=0.001)
 
         self.assertListEqual(rx_callback, [1])
@@ -741,15 +741,15 @@ class TestSleep(BlockingTestCaseBase):
         connection = self._connect()
 
         # Try with duration=0
-        start_time = time.time()
+        start_time = time_now()
         connection.sleep(duration=0)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertLess(elapsed, 0.25)
 
         # Try with duration=0.005
-        start_time = time.time()
+        start_time = time_now()
         connection.sleep(duration=0.005)
-        elapsed = time.time() - start_time
+        elapsed = time_now() - start_time
         self.assertGreaterEqual(elapsed, 0.005)
         self.assertLess(elapsed, 0.25)
 
diff --git a/tests/acceptance/io_services_tests.py b/tests/acceptance/io_services_tests.py
index 6b342eb..897e356 100644
--- a/tests/acceptance/io_services_tests.py
+++ b/tests/acceptance/io_services_tests.py
@@ -145,10 +145,10 @@ class TestCallLaterDoesNotCallAheadOfTime(AsyncServicesTestBase,
             loop.stop()
             bucket.append('I was here')
 
-        start_time = time.time()
+        start_time = pika.compat.time_now()
         loop.call_later(0.1, callback)
         loop.run()
-        self.assertGreaterEqual(round(time.time() - start_time, 3), 0.1)
+        self.assertGreaterEqual(round(pika.compat.time_now() - start_time, 3), 0.1)
         self.assertEqual(bucket, ['I was here'])
 
 
diff --git a/tests/acceptance/test_utils.py b/tests/acceptance/test_utils.py
index adf5934..b54c702 100644
--- a/tests/acceptance/test_utils.py
+++ b/tests/acceptance/test_utils.py
@@ -4,6 +4,7 @@ import functools
 import logging
 import time
 import traceback
+import pika.compat
 
 
 def retry_assertion(timeout_sec, retry_interval_sec=0.1):
@@ -36,7 +37,7 @@ def retry_assertion(timeout_sec, retry_interval_sec=0.1):
             """The wrapper"""
 
             num_attempts = 0
-            start_time = time.time()
+            start_time = pika.compat.time_now()
 
             while True:
                 num_attempts += 1
@@ -45,7 +46,7 @@ def retry_assertion(timeout_sec, retry_interval_sec=0.1):
                     result = func(*args, **kwargs)
                 except AssertionError:
 
-                    now = time.time()
+                    now = pika.compat.time_now()
                     # Compensate for time adjustment
                     if now < start_time:
                         start_time = now
diff --git a/tests/unit/select_connection_ioloop_tests.py b/tests/unit/select_connection_ioloop_tests.py
index c24b72f..9dd2c1f 100644
--- a/tests/unit/select_connection_ioloop_tests.py
+++ b/tests/unit/select_connection_ioloop_tests.py
@@ -657,12 +657,12 @@ class SelectPollerTestPollWithoutSockets(unittest.TestCase):
 
         delay = poller._get_wait_seconds()
         self.assertIsNotNone(delay)
-        deadline = time.time() + delay
+        deadline = pika.compat.time_now() + delay
 
         while True:
             poller._process_timeouts()
 
-            if time.time() < deadline:
+            if pika.compat.time_now() < deadline:
                 self.assertEqual(timer_call_container, [])
             else:
                 # One last time in case deadline reached after previous
diff --git a/tests/unit/select_connection_timer_tests.py b/tests/unit/select_connection_timer_tests.py
index 13cae53..8b2e4ae 100644
--- a/tests/unit/select_connection_timer_tests.py
+++ b/tests/unit/select_connection_timer_tests.py
@@ -4,11 +4,13 @@ Tests for SelectConnection _Timer and _Timeout classes
 
 """
 
+import math
 import time
 import unittest
 
 import mock
 
+import pika.compat
 from pika.adapters import select_connection
 
 
@@ -22,6 +24,16 @@ from pika.adapters import select_connection
 # pylint: disable=C0103
 
 
+def test_now():
+    # pika/pika#1184
+    # Note that time is a float, and these tests depend
+    # on exact math. Round up the value to ensure that
+    # CI doesn't fail because of something like this:
+    # raise self.failureException('6.000000000000028 != 6')
+    # https://travis-ci.org/pika/pika/jobs/489828602
+    return math.ceil(pika.compat.time_now())
+
+
 class ChildTimeout(select_connection._Timeout):
 
     def __init__(self, *args, **kwargs):
@@ -39,7 +51,7 @@ class TimeoutClassTests(unittest.TestCase):
     """Test select_connection._Timeout class"""
 
     def test_properties(self):
-        now = time.time()
+        now = test_now()
         cb = lambda: None
         timeout = select_connection._Timeout(now + 5.3, cb)
         self.assertIs(timeout.callback, cb)
@@ -270,17 +282,17 @@ class TimerClassTests(unittest.TestCase):
         self.assertIsNone(timer.get_remaining_interval())
 
     def test_call_later_non_negative_delay_check(self):
-        now = time.time()
+        now = test_now()
 
         # 0 delay is okay
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             timer = select_connection._Timer()
             timer.call_later(0, lambda: None)
             self.assertEqual(timer._timeout_heap[0].deadline, now)
             self.assertEqual(timer.get_remaining_interval(), 0)
 
         # Positive delay is okay
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             timer = select_connection._Timer()
             timer.call_later(0.5, lambda: None)
             self.assertEqual(timer._timeout_heap[0].deadline, now + 0.5)
@@ -294,9 +306,9 @@ class TimerClassTests(unittest.TestCase):
                       cm.exception.args[0])
 
     def test_call_later_single_timer_expires(self):
-        now = time.time()
+        now = test_now()
 
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             bucket = []
             timer = select_connection._Timer()
             timer.call_later(5, lambda: bucket.append(1))
@@ -307,7 +319,7 @@ class TimerClassTests(unittest.TestCase):
             self.assertEqual(timer.get_remaining_interval(), 5)
 
         # Advance time by 5 seconds and expect the timer to expire
-        with mock.patch('time.time', return_value=now + 5):
+        with mock.patch('pika.compat.time_now', return_value=now + 5):
             self.assertEqual(timer.get_remaining_interval(), 0)
             timer.process_timeouts()
             self.assertEqual(bucket, [1])
@@ -315,12 +327,12 @@ class TimerClassTests(unittest.TestCase):
             self.assertIsNone(timer.get_remaining_interval())
 
     def test_call_later_multiple_timers(self):
-        now = time.time()
+        now = test_now()
 
         bucket = []
         timer = select_connection._Timer()
 
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             timer.call_later(5, lambda: bucket.append(1))
             timer.call_later(5, lambda: bucket.append(2))
             timer.call_later(10, lambda: bucket.append(3))
@@ -332,7 +344,7 @@ class TimerClassTests(unittest.TestCase):
             self.assertEqual(timer.get_remaining_interval(), 5)
 
         # Advance time by 6 seconds and expect first two timers to expire
-        with mock.patch('time.time', return_value=now + 6):
+        with mock.patch('pika.compat.time_now', return_value=now + 6):
             self.assertEqual(timer.get_remaining_interval(), 0)
             timer.process_timeouts()
             self.assertEqual(bucket, [1, 2])
@@ -340,7 +352,7 @@ class TimerClassTests(unittest.TestCase):
             self.assertEqual(timer.get_remaining_interval(), 4)
 
         # Advance time by 10 seconds and expect the 3rd timeout to expire
-        with mock.patch('time.time', return_value=now + 10):
+        with mock.patch('pika.compat.time_now', return_value=now + 10):
             self.assertEqual(timer.get_remaining_interval(), 0)
             timer.process_timeouts()
             self.assertEqual(bucket, [1, 2, 3])
@@ -348,12 +360,12 @@ class TimerClassTests(unittest.TestCase):
             self.assertIsNone(timer.get_remaining_interval())
 
     def test_add_and_remove_timeout(self):
-        now = time.time()
+        now = test_now()
 
         bucket = []
         timer = select_connection._Timer()
 
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             timer.call_later(10, lambda: bucket.append(3)) # t3
             t2 = timer.call_later(6, lambda: bucket.append(2))
             t1 = timer.call_later(5, lambda: bucket.append(1))
@@ -380,7 +392,7 @@ class TimerClassTests(unittest.TestCase):
 
         # Advance time by 6 seconds to expire t1 and t2 and verify they don't
         # fire
-        with mock.patch('time.time', return_value=now + 6):
+        with mock.patch('pika.compat.time_now', return_value=now + 6):
             self.assertEqual(timer.get_remaining_interval(), 0)
             timer.process_timeouts()
             self.assertEqual(bucket, [])
@@ -389,7 +401,7 @@ class TimerClassTests(unittest.TestCase):
             self.assertEqual(timer.get_remaining_interval(), 4)
 
         # Advance time by 10 seconds to expire t3 and verify it fires
-        with mock.patch('time.time', return_value=now + 10):
+        with mock.patch('pika.compat.time_now', return_value=now + 10):
             self.assertEqual(timer.get_remaining_interval(), 0)
             timer.process_timeouts()
             self.assertEqual(bucket, [3])
@@ -397,13 +409,13 @@ class TimerClassTests(unittest.TestCase):
             self.assertIsNone(timer.get_remaining_interval())
 
     def test_gc_of_unexpired_timeouts(self):
-        now = time.time()
+        now = test_now()
         bucket = []
         timer = select_connection._Timer()
 
         with mock.patch.multiple(select_connection._Timer,
                                  _GC_CANCELLATION_THRESHOLD=1):
-            with mock.patch('time.time', return_value=now):
+            with mock.patch('pika.compat.time_now', return_value=now):
                 t3 = timer.call_later(10, lambda: bucket.append(3))
                 t2 = timer.call_later(6, lambda: bucket.append(2))
                 t1 = timer.call_later(5, lambda: bucket.append(1))
@@ -430,11 +442,11 @@ class TimerClassTests(unittest.TestCase):
                 self.assertEqual(timer._num_cancellations, 0)
 
     def test_add_timeout_from_another_timeout(self):
-        now = time.time()
+        now = test_now()
         bucket = []
         timer = select_connection._Timer()
 
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             t1 = timer.call_later(
                 5,
                 lambda: bucket.append(
@@ -443,7 +455,7 @@ class TimerClassTests(unittest.TestCase):
         # Advance time by 10 seconds and verify that t1 fires and creates t2,
         # but timer manager defers firing of t2 to next `process_timeouts` in
         # order to avoid IO starvation
-        with mock.patch('time.time', return_value=now + 10):
+        with mock.patch('pika.compat.time_now', return_value=now + 10):
             timer.process_timeouts()
             t2 = bucket.pop()
             self.assertIsInstance(t2, select_connection._Timeout)
@@ -459,11 +471,11 @@ class TimerClassTests(unittest.TestCase):
             self.assertEqual(timer.get_remaining_interval(), None)
 
     def test_cancel_unexpired_timeout_from_another_timeout(self):
-        now = time.time()
+        now = test_now()
         bucket = []
         timer = select_connection._Timer()
 
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             t2 = timer.call_later(10, lambda: bucket.append(2))
             t1 = timer.call_later(5, lambda: timer.remove_timeout(t2))
 
@@ -471,7 +483,7 @@ class TimerClassTests(unittest.TestCase):
 
         # Advance time by 6 seconds and check that t2 is cancelled, but not
         # removed from timeout heap
-        with mock.patch('time.time', return_value=now + 6):
+        with mock.patch('pika.compat.time_now', return_value=now + 6):
             timer.process_timeouts()
             self.assertIsNone(t2.callback)
             self.assertEqual(timer.get_remaining_interval(), 4)
@@ -480,7 +492,7 @@ class TimerClassTests(unittest.TestCase):
 
         # Advance time by 10 seconds and verify that t2 is removed without
         # firing
-        with mock.patch('time.time', return_value=now + 10):
+        with mock.patch('pika.compat.time_now', return_value=now + 10):
             timer.process_timeouts()
             self.assertEqual(bucket, [])
             self.assertIsNone(timer.get_remaining_interval())
@@ -489,11 +501,11 @@ class TimerClassTests(unittest.TestCase):
 
 
     def test_cancel_expired_timeout_from_another_timeout(self):
-        now = time.time()
+        now = test_now()
         bucket = []
         timer = select_connection._Timer()
 
-        with mock.patch('time.time', return_value=now):
+        with mock.patch('pika.compat.time_now', return_value=now):
             t2 = timer.call_later(10, lambda: bucket.append(2))
             t1 = timer.call_later(
                 5,
@@ -504,7 +516,7 @@ class TimerClassTests(unittest.TestCase):
 
         # Advance time by 10 seconds and check that t2 is cancelled and
         # removed from timeout heap
-        with mock.patch('time.time', return_value=now + 10):
+        with mock.patch('pika.compat.time_now', return_value=now + 10):
             timer.process_timeouts()
             self.assertEqual(bucket, [])
             self.assertIsNone(t2.callback)