91% Coverage for celery.worker.heartbeat

celery/tests/test_worker_heartbeat.py

@@ -1,21 +1,21 @@
-from celery.tests.utils import unittest
-
 from celery.worker.heartbeat import Heart
 
+from celery.tests.utils import unittest, sleepdeprived
+
 
 class MockDispatcher(object):
-    shutdown_event = None
-    next_iter = False
+    heart = None
+    next_iter = 0
 
     def __init__(self):
         self.sent = []
 
     def send(self, msg):
         self.sent.append(msg)
-        if self.shutdown_event:
-            if self.next_iter:
-                self.shutdown_event.set()
-            self.next_iter = True
+        if self.heart:
+            if self.next_iter > 10:
+                self.heart._shutdown.set()
+            self.next_iter += 1
 
 
 class MockDispatcherRaising(object):
@@ -51,13 +51,35 @@ class TestHeart(unittest.TestCase):
         h2.stop()
         self.assertFalse(h2._joined)
 
+    def test_time_raises_TypeError(self):
+        from celery.worker import heartbeat
+
+        def raises_TypeError(exc):
+            raise TypeError("x")
+
+        prev_time, heartbeat.time = heartbeat.time, raises_TypeError
+        try:
+            eventer = MockDispatcher()
+            heart = Heart(eventer, interval=0.1)
+            heart.run()
+            self.assertIn("worker-online", eventer.sent)
+            self.assertNotIn("worker-heartbeat", eventer.sent)
+
+        finally:
+            heartbeat.time = prev_time
+
+    @sleepdeprived
     def test_run_manages_cycle(self):
         eventer = MockDispatcher()
-        heart = Heart(eventer, interval=1)
-        eventer.shutdown_event = heart._shutdown
+        heart = Heart(eventer, interval=0.1)
+        eventer.heart = heart
         heart.run()
         self.assertEqual(heart._state, "RUN")
         self.assertTrue(heart._shutdown.isSet())
+        heart._shutdown.clear()
+        heart._stopped.clear()
+        eventer.next_iter = 0
+        heart.run()
 
     def test_run(self):
         eventer = MockDispatcher()

celery/worker/heartbeat.py

@@ -39,9 +39,9 @@ class Heart(threading.Thread):
         while 1:
             try:
                 now = time()
-            except TypeError:  # pragma: no cover
+            except TypeError:
                 # we lost the race at interpreter shutdown,
-                # so time has been collected by gc.
+                # so time() has been collected by gc.
                 return
 
             if not last_beat or now > last_beat + (60.0 / bpm):