Fixes test failures. Closes #551

celery/__init__.py

@@ -14,7 +14,6 @@ __docformat__ = "restructuredtext"
 
 # -eof meta-
 
-import os
 import sys
 
 if sys.version_info < (2, 5):

celery/bin/base.py

@@ -110,6 +110,8 @@ class Command(object):
             sys.stderr.write(
                 "\nUnrecognized command line arguments: %s\n" % (
                     ", ".join(args), ))
+            import traceback
+            traceback.print_stack(file=sys.stderr)
             sys.stderr.write("\nTry --help?\n")
             sys.exit(1)
         return self.run(*args, **vars(options))

celery/tests/test_backends/test_redis_unit.py

@@ -44,10 +44,20 @@ class Redis(object):
     def delete(self, key):
         self.keyspace.pop(key)
 
+    def publish(self, key, value):
+        pass
+
 
 class redis(object):
     Redis = Redis
 
+    class ConnectionPool(object):
+
+        def __init__(self, **kwargs):
+            pass
+
+
+
 
 class test_RedisBackend(unittest.TestCase):
 

celery/tests/test_bin/test_celeryd.py

@@ -253,19 +253,19 @@ class test_Worker(AppCase):
             raise SkipTest("Not applicable on Windows")
         warnings.resetwarnings()
 
-        def geteuid():
+        def getuid():
             return 0
 
-        prev, os.geteuid = os.geteuid, geteuid
+        prev, os.getuid = os.getuid, getuid
         try:
             with catch_warnings(record=True) as log:
                 worker = self.Worker()
                 worker.run()
                 self.assertTrue(log)
-                self.assertIn("superuser privileges is not encouraged",
+                self.assertIn("superuser privileges is discouraged",
                               log[0].message.args[0])
         finally:
-            os.geteuid = prev
+            os.getuid = prev
 
     @disable_stdouts
     def test_use_pidfile(self):

celery/tests/test_worker/test_worker_autoscale.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 
 import logging
+import sys
 
 from time import time
 
@@ -53,7 +54,7 @@ class test_Autoscaler(unittest.TestCase):
             alive = True
             joined = False
 
-            def isAlive(self):
+            def is_alive(self):
                 return self.alive
 
             def join(self, timeout=None):
@@ -90,7 +91,7 @@ class test_Autoscaler(unittest.TestCase):
         class Scaler(autoscale.Autoscaler):
             scale_called = False
 
-            def scale(self):
+            def body(self):
                 self.scale_called = True
                 self._is_shutdown.set()
 
@@ -140,12 +141,16 @@ class test_Autoscaler(unittest.TestCase):
 
         class _Autoscaler(autoscale.Autoscaler):
 
-            def scale(self):
+            def body(self):
                 self._is_shutdown.set()
                 raise OSError("foo")
-
         x = _Autoscaler(self.pool, 10, 3, logger=logger)
-        x.logger = Mock()
-        x.run()
+
+        stderr = Mock()
+        p, sys.stderr = sys.stderr, stderr
+        try:
+            x.run()
+        finally:
+            sys.stderr = p
         _exit.assert_called_with(1)
-        self.assertTrue(x.logger.error.call_count)
+        self.assertTrue(stderr.write.call_count)

celery/tests/test_worker/test_worker_mediator.py

@@ -1,5 +1,7 @@
 from __future__ import absolute_import
 
+import sys
+
 from Queue import Queue
 
 from mock import Mock, patch
@@ -40,7 +42,7 @@ class test_Mediator(unittest.TestCase):
         self.assertTrue(m._is_shutdown.isSet())
         self.assertTrue(m._is_stopped.isSet())
 
-    def test_mediator_move(self):
+    def test_mediator_body(self):
         ready_queue = Queue()
         got = {}
 
@@ -50,7 +52,7 @@ class test_Mediator(unittest.TestCase):
         m = Mediator(ready_queue, mycallback)
         ready_queue.put(MockTask("George Costanza"))
 
-        m.move()
+        m.body()
 
         self.assertEqual(got["value"], "George Costanza")
 
@@ -60,7 +62,7 @@ class test_Mediator(unittest.TestCase):
 
         class _Mediator(Mediator):
 
-            def move(self):
+            def body(self):
                 try:
                     raise KeyError("foo")
                 finally:
@@ -69,11 +71,17 @@ class test_Mediator(unittest.TestCase):
         ready_queue = Queue()
         ms[0] = m = _Mediator(ready_queue, None)
         ready_queue.put(MockTask("George Constanza"))
-        m.run()
 
+        stderr = Mock()
+        p, sys.stderr = sys.stderr, stderr
+        try:
+            m.run()
+        finally:
+            sys.stderr = p
         self.assertTrue(_exit.call_count)
+        self.assertTrue(stderr.write.call_count)
 
-    def test_mediator_move_exception(self):
+    def test_mediator_body_exception(self):
         ready_queue = Queue()
 
         def mycallback(value):
@@ -82,7 +90,7 @@ class test_Mediator(unittest.TestCase):
         m = Mediator(ready_queue, mycallback)
         ready_queue.put(MockTask("Elaine M. Benes"))
 
-        m.move()
+        m.body()
 
     def test_run(self):
         ready_queue = Queue()
@@ -100,7 +108,7 @@ class test_Mediator(unittest.TestCase):
         self.assertTrue(m._is_shutdown.isSet())
         self.assertTrue(m._is_stopped.isSet())
 
-    def test_mediator_move_revoked(self):
+    def test_mediator_body_revoked(self):
         ready_queue = Queue()
         got = {}
 
@@ -113,7 +121,7 @@ class test_Mediator(unittest.TestCase):
         revoked_tasks.add(t.task_id)
         ready_queue.put(t)
 
-        m.move()
+        m.body()
 
         self.assertNotIn("value", got)
         self.assertTrue(t.on_ack.call_count)

celery/utils/threads.py

@@ -1,3 +1,5 @@
+from __future__ import absolute_import
+
 import os
 import sys
 import threading
@@ -33,18 +35,17 @@ class bgThread(Thread):
         self.daemon = True
         self.name = name or self.__class__.__name__
 
-    def next(self):
+    def body(self):
         raise NotImplementedError("subclass responsibility")
-    __next__ = next  # 2to3.
 
     def on_crash(self, msg, *fmt, **kwargs):
-        sys.stderr.write(msg + "\n" % fmt)
+        sys.stderr.write((msg + "\n") % fmt)
 
     def run(self):
         shutdown = self._is_shutdown
         while not shutdown.is_set():
             try:
-                self.next()
+                self.body()
             except Exception, exc:
                 self.on_crash("%r crashed: %r", self.name, exc, exc_info=True)
                 # exiting by normal means does not work here, so force exit.
@@ -61,5 +62,6 @@ class bgThread(Thread):
         """Graceful shutdown."""
         self._is_shutdown.set()
         self._is_stopped.wait()
-        if self.is_alive:
+        if self.is_alive():
+            print("JOINING")
             self.join(1e100)

celery/worker/autoscale.py

@@ -42,7 +42,7 @@ class Autoscaler(bgThread):
 
         assert self.keepalive, "can't scale down too fast."
 
-    def next(self):
+    def body(self):
         with self.mutex:
             current = min(self.qty, self.max_concurrency)
             if current > self.processes:
@@ -51,7 +51,7 @@ class Autoscaler(bgThread):
                 self.scale_down(
                     (self.processes - current) - self.min_concurrency)
         sleep(1.0)
-    scale = next  # XXX compat
+    scale = body  # XXX compat
 
     def update(self, max=None, min=None):
         with self.mutex:

celery/worker/mediator.py

@@ -18,6 +18,7 @@
 """
 from __future__ import absolute_import
 
+import sys
 import traceback
 
 from Queue import Empty
@@ -41,7 +42,7 @@ class Mediator(bgThread):
         self.callback = callback
         super(Mediator, self).__init__()
 
-    def next(self):
+    def body(self):
         try:
             task = self.ready_queue.get(timeout=1.0)
         except Empty:
@@ -63,4 +64,4 @@ class Mediator(bgThread):
                               extra={"data": {"id": task.task_id,
                                               "name": task.task_name,
                                               "hostname": task.hostname}})
-    move = next   # XXX compat
+    move = body   # XXX compat