83% coverage

celery/beat.py

@@ -1,5 +1,4 @@
 import time
-import math
 import shelve
 import threading
 from datetime import datetime

celery/tests/test_utils_info.py

@@ -0,0 +1,63 @@
+import unittest
+
+from celery.utils import info
+
+RANDTEXT = """\
+The quick brown
+fox jumps
+over the
+lazy dog\
+"""
+
+RANDTEXT_RES = """\
+    The quick brown
+    fox jumps
+    over the
+    lazy dog\
+"""
+
+ROUTE = {"queue1": {
+            "exchange": "exchange1",
+            "exchange_type": "type1",
+            "binding_key": "bind1"},
+         "queue2": {
+            "exchange": "exchange2",
+            "exchange_type": "type2",
+            "binding_key": "bind2"}}
+
+
+ROUTE_FORMAT = """
+. queue1 -> exchange:exchange1 (type1) binding:bind1
+. queue2 -> exchange:exchange2 (type2) binding:bind2
+""".strip()
+
+
+
+class TestInfo(unittest.TestCase):
+
+    def test_humanize_seconds(self):
+        t = ((4 * 60 * 60 * 24, "4 days"),
+             (1 * 60 * 60 * 24, "1 day"),
+             (4 * 60 * 60, "4 hours"),
+             (1 * 60 * 60, "1 hour"),
+             (4 * 60, "4 minutes"),
+             (1 * 60, "1 minute"),
+             (4, "4.00 seconds"),
+             (1, "1.00 second"),
+             (4.3567631221, "4.36 seconds"),
+             (0, "now"))
+
+        for seconds, human in t:
+            self.assertEquals(info.humanize_seconds(seconds), human)
+
+        self.assertEquals(info.humanize_seconds(4, prefix="about "),
+                          "about 4.00 seconds")
+
+    def test_textindent(self):
+        self.assertEquals(info.textindent(RANDTEXT, 4), RANDTEXT_RES)
+
+    def test_format_routing_table(self):
+        self.assertEquals(info.format_routing_table(ROUTE), ROUTE_FORMAT)
+
+    def test_broker_info(self):
+        info.format_broker_info()

celery/tests/test_worker_controllers.py

@@ -2,17 +2,22 @@ import time
 import unittest
 from Queue import Queue
 
+from celery.utils import gen_unique_id
 from celery.worker.controllers import Mediator
-from celery.worker.controllers import BackgroundThread
+from celery.worker.controllers import BackgroundThread, ScheduleController
 
 
 class MockTask(object):
     task_id = 1234
     task_name = "mocktask"
+    acked = False
 
     def __init__(self, value, **kwargs):
         self.value = value
 
+    def on_ack(self):
+        self.acked = True
+
 
 class MyBackgroundThread(BackgroundThread):
 
@@ -68,3 +73,45 @@ class TestMediator(unittest.TestCase):
         m.on_iteration()
 
         self.assertEquals(got["value"], "George Constanza")
+
+    def test_mediator_on_iteration_revoked(self):
+        ready_queue = Queue()
+        got = {}
+
+        def mycallback(value):
+            got["value"] = value.value
+
+        m = Mediator(ready_queue, mycallback)
+        t = MockTask("Jerry Seinfeld")
+        t.task_id = gen_unique_id()
+        from celery.worker.revoke import revoked
+        revoked.add(t.task_id)
+        ready_queue.put(t)
+
+        m.on_iteration()
+
+        self.assertTrue("value" not in got)
+        self.assertTrue(t.acked)
+
+
+class TestScheduleController(unittest.TestCase):
+
+    def test_on_iteration(self):
+        times = range(10) + [None]
+        c = ScheduleController(times)
+
+        import time
+        slept = [None]
+
+        def _sleep(count):
+            slept[0] = count
+
+        old_sleep = time.sleep
+        time.sleep = _sleep
+        try:
+            for i in times:
+                c.on_iteration()
+                res = i is None and 1 or i
+                self.assertEquals(slept[0], res)
+        finally:
+            time.sleep = old_sleep

celery/tests/test_worker_heartbeat.py

@@ -0,0 +1,51 @@
+import unittest
+
+from celery.worker.heartbeat import Heart
+
+
+class MockDispatcher(object):
+
+    def __init__(self):
+        self.sent = []
+
+    def send(self, msg):
+        self.sent.append(msg)
+
+
+class MockDispatcherRaising(object):
+
+    def send(self, msg):
+        if msg == "worker-offline":
+            raise Exception("foo")
+
+
+class TestHeart(unittest.TestCase):
+
+    def test_run(self):
+        eventer = MockDispatcher()
+
+        heart = Heart(eventer, interval=1)
+        heart._shutdown.set()
+        heart.run()
+        self.assertTrue(heart._state == "RUN")
+        self.assertTrue("worker-online" in eventer.sent)
+        self.assertTrue("worker-heartbeat" in eventer.sent)
+        self.assertTrue("worker-offline" in eventer.sent)
+
+        self.assertTrue(heart._stopped.isSet())
+
+        heart.stop()
+        heart.stop()
+        self.assertTrue(heart._state == "CLOSE")
+ 
+        heart = Heart(eventer, interval=0.00001)
+        heart._shutdown.set()
+        for i in range(10):
+            heart.run()
+
+    def test_run_stopped_is_set_even_if_send_breaks(self):
+        eventer = MockDispatcherRaising()
+        heart = Heart(eventer, interval=1)
+        heart._shutdown.set()
+        self.assertRaises(Exception, heart.run)
+        self.assertTrue(heart._stopped.isSet())

celery/utils/info.py

@@ -1,3 +1,5 @@
+import math
+
 from celery import conf
 from celery.messaging import establish_connection
 
@@ -10,7 +12,7 @@ BROKER_FORMAT = "%(carrot_backend)s://%(userid)s@%(host)s%(port)s%(vhost)s"
 TIME_UNITS = (("day", 60 * 60 * 24, lambda n: int(math.ceil(n))),
               ("hour", 60 * 60, lambda n: int(math.ceil(n))),
               ("minute", 60, lambda n: int(math.ceil(n))),
-              ("second", 1, lambda n: "%.2d" % n))
+              ("second", 1, lambda n: "%.2f" % n))
 
 
 def humanize_seconds(secs, prefix=""):

celery/worker/heartbeat.py

@@ -38,12 +38,12 @@ class Heart(threading.Thread):
 
         last_beat = None
         while 1:
-            if self._shutdown.isSet():
-                break
             now = time()
             if not last_beat or now > last_beat + (60.0 / bpm):
                 last_beat = now
                 dispatch("worker-heartbeat")
+            if self._shutdown.isSet():
+                break
             sleep(1)
 
         try: