Tests now passes on Python 2.5

celery/tests/compat.py

@@ -0,0 +1,63 @@
+import sys
+from warnings import WarningMessage
+
+
+class catch_warnings(object):
+
+    """A context manager that copies and restores the warnings filter upon
+    exiting the context.
+
+    The 'record' argument specifies whether warnings should be captured by a
+    custom implementation of warnings.showwarning() and be appended to a list
+    returned by the context manager. Otherwise None is returned by the context
+    manager. The objects appended to the list are arguments whose attributes
+    mirror the arguments to showwarning().
+
+    The 'module' argument is to specify an alternative module to the module
+    named 'warnings' and imported under that name. This argument is only
+    useful when testing the warnings module itself.
+
+    """
+
+    def __init__(self, record=False, module=None):
+        """Specify whether to record warnings and if an alternative module
+        should be used other than sys.modules['warnings'].
+
+        For compatibility with Python 3.0, please consider all arguments to be
+        keyword-only.
+
+        """
+        self._record = record
+        self._module = sys.modules['warnings'] if module is None else module
+        self._entered = False
+
+    def __repr__(self):
+        args = []
+        if self._record:
+            args.append("record=True")
+        if self._module is not sys.modules['warnings']:
+            args.append("module=%r" % self._module)
+        name = type(self).__name__
+        return "%s(%s)" % (name, ", ".join(args))
+
+    def __enter__(self):
+        if self._entered:
+            raise RuntimeError("Cannot enter %r twice" % self)
+        self._entered = True
+        self._filters = self._module.filters
+        self._module.filters = self._filters[:]
+        self._showwarning = self._module.showwarning
+        if self._record:
+            log = []
+            def showwarning(*args, **kwargs):
+                log.append(WarningMessage(*args, **kwargs))
+            self._module.showwarning = showwarning
+            return log
+        else:
+            return None
+
+    def __exit__(self, *exc_info):
+        if not self._entered:
+            raise RuntimeError("Cannot exit %r without entering first" % self)
+        self._module.filters = self._filters
+        self._module.showwarning = self._showwarning

celery/tests/test_backends/test_cache.py

@@ -2,6 +2,7 @@ import sys
 import unittest
 
 from billiard.serialization import pickle
+from django.core.cache.backends.base import InvalidCacheBackendError
 
 from celery import result
 from celery import states
@@ -100,8 +101,13 @@ class TestMemcacheWrapper(unittest.TestCase):
 
     def test_memcache_wrapper(self):
 
-        from django.core.cache.backends import memcached
-        from django.core.cache.backends import locmem
+        try:
+            from django.core.cache.backends import memcached
+            from django.core.cache.backends import locmem
+        except InvalidCacheBackendError:
+            sys.stderr.write(
+                "\n* Memcache library is not installed. Skipping test.\n")
+            return
         prev_cache_cls = memcached.CacheClass
         memcached.CacheClass = locmem.CacheClass
         prev_backend_module = sys.modules.pop("celery.backends.cache")

celery/tests/test_buckets.py

@@ -101,13 +101,8 @@ class TestRateLimitString(unittest.TestCase):
                           100 / 60.0)
         self.assertEquals(buckets.parse_ratelimit_string("10/h"),
                           10 / 60.0 / 60.0)
-        self.assertEquals(buckets.parse_ratelimit_string("0xffec/s"), 0xffec)
-        self.assertEquals(buckets.parse_ratelimit_string("0xcda/m"),
-                          0xcda / 60.0)
-        self.assertEquals(buckets.parse_ratelimit_string("0xF/h"),
-                          0xf / 60.0 / 60.0)
 
-        for zero in ("0x0", "0b0", "0o0", 0, None, "0/m", "0/h", "0/s"):
+        for zero in (0, None, "0", "0/m", "0/h", "0/s"):
             self.assertEquals(buckets.parse_ratelimit_string(zero), 0)
 
 

celery/tests/test_worker.py

@@ -11,6 +11,7 @@ from billiard.serialization import pickle
 
 from celery import conf
 from celery.utils import gen_unique_id, noop
+from celery.tests.compat import catch_warnings
 from celery.worker import WorkController
 from celery.worker.listener import CarrotListener, RUN, CLOSE
 from celery.worker.job import TaskWrapper
@@ -221,8 +222,7 @@ class TestCarrotListener(unittest.TestCase):
         m = create_message(backend, unknown={"baz": "!!!"})
         l.event_dispatcher = MockEventDispatcher()
         l.control_dispatch = MockControlDispatch()
-        import warnings
-        with warnings.catch_warnings(record=True) as log:
+        with catch_warnings(record=True) as log:
                 l.receive_message(m.decode(), m)
                 self.assertTrue(log)
                 self.assertTrue("unknown message" in log[0].message.args[0])

celery/tests/test_worker_job.py

@@ -14,6 +14,7 @@ from celery import states
 from celery.log import setup_logger
 from celery.task.base import Task
 from celery.utils import gen_unique_id
+from celery.tests.compat import catch_warnings
 from celery.models import TaskMeta
 from celery.result import AsyncResult
 from celery.worker.job import WorkerTaskTrace, TaskWrapper
@@ -225,8 +226,7 @@ class TestTaskWrapper(unittest.TestCase):
 
         WorkerTaskTrace.execute = _error_exec
         try:
-            import warnings
-            with warnings.catch_warnings(record=True) as log:
+            with catch_warnings(record=True) as log:
                 res = execute_and_trace(mytask.name, gen_unique_id(),
                                         [4], {})
                 self.assertTrue(isinstance(res, ExceptionInfo))

celery/worker/buckets.py

@@ -8,7 +8,6 @@ from celery.utils import all
 RATE_MODIFIER_MAP = {"s": lambda n: n,
                      "m": lambda n: n / 60.0,
                      "h": lambda n: n / 60.0 / 60.0}
-BASE_IDENTIFIERS = {"0x": 16, "0o": 8, "0b": 2}
 
 
 class RateLimitExceeded(Exception):
@@ -25,12 +24,8 @@ def parse_ratelimit_string(rate_limit):
 
     if rate_limit:
         if isinstance(rate_limit, basestring):
-            base = BASE_IDENTIFIERS.get(rate_limit[:2], 10)
-            try:
-                return int(rate_limit, base)
-            except ValueError:
-                ops, _, modifier = partition(rate_limit, "/")
-                return RATE_MODIFIER_MAP[modifier](int(ops, base)) or 0
+            ops, _, modifier = partition(rate_limit, "/")
+            return RATE_MODIFIER_MAP[modifier or "s"](int(ops)) or 0
         return rate_limit or 0
     return 0