100% coverage for celery.utils.functional

celery/datastructures.py

@@ -507,14 +507,21 @@ class ConfigurationView(AttributeDictMixin):
         return (self[key] for key in self)
     itervalues = _iterate_values
 
-    def keys(self):
-        return list(self._iterate_keys())
+    if sys.version_info[0] == 3:  # pragma: no cover
+        keys = _iterate_keys
+        items = _iterate_items
+        values = _iterate_values
+
+    else:  # noqa
+        def keys(self):
+            return list(self._iterate_keys())
+
+        def items(self):
+            return list(self._iterate_items())
 
-    def items(self):
-        return list(self._iterate_items())
+        def values(self):
+            return list(self._iterate_values())
 
-    def values(self):
-        return list(self._iterate_values())
 MutableMapping.register(ConfigurationView)
 
 

celery/tests/utilities/test_datastructures.py

@@ -2,14 +2,13 @@ from __future__ import absolute_import
 
 from celery.datastructures import (
     ExceptionInfo,
-    LRUCache,
     LimitedSet,
     AttributeDict,
     DictAttribute,
     ConfigurationView,
     DependencyGraph,
 )
-from celery.five import THREAD_TIMEOUT_MAX, items, range
+from celery.five import items
 from celery.tests.utils import Case, WhateverIO
 
 
@@ -180,92 +179,6 @@ class test_LimitedSet(Case):
         self.assertIsInstance(s.as_dict(), dict)
 
 
-class test_LRUCache(Case):
-
-    def test_expires(self):
-        limit = 100
-        x = LRUCache(limit=limit)
-        slots = list(range(limit * 2))
-        for i in slots:
-            x[i] = i
-        self.assertListEqual(list(x.keys()), list(slots[limit:]))
-
-    def test_update_expires(self):
-        limit = 100
-        x = LRUCache(limit=limit)
-        slots = list(range(limit * 2))
-        for i in slots:
-            x.update({i: i})
-
-        self.assertListEqual(list(x.keys()), list(slots[limit:]))
-
-    def test_least_recently_used(self):
-        x = LRUCache(3)
-
-        x[1], x[2], x[3] = 1, 2, 3
-        self.assertEqual(list(x.keys()), [1, 2, 3])
-
-        x[4], x[5] = 4, 5
-        self.assertEqual(list(x.keys()), [3, 4, 5])
-
-        # access 3, which makes it the last used key.
-        x[3]
-        x[6] = 6
-        self.assertEqual(list(x.keys()), [5, 3, 6])
-
-        x[7] = 7
-        self.assertEqual(list(x.keys()), [3, 6, 7])
-
-    def assertSafeIter(self, method, interval=0.01, size=10000):
-        from threading import Thread, Event
-        from time import sleep
-        x = LRUCache(size)
-        x.update(zip(range(size), range(size)))
-
-        class Burglar(Thread):
-
-            def __init__(self, cache):
-                self.cache = cache
-                self._is_shutdown = Event()
-                self._is_stopped = Event()
-                Thread.__init__(self)
-
-            def run(self):
-                while not self._is_shutdown.isSet():
-                    try:
-                        self.cache.data.popitem(last=False)
-                    except KeyError:
-                        break
-                self._is_stopped.set()
-
-            def stop(self):
-                self._is_shutdown.set()
-                self._is_stopped.wait()
-                self.join(THREAD_TIMEOUT_MAX)
-
-        burglar = Burglar(x)
-        burglar.start()
-        try:
-            for _ in getattr(x, method)():
-                sleep(0.0001)
-        finally:
-            burglar.stop()
-
-    def test_safe_to_remove_while_iteritems(self):
-        self.assertSafeIter('iteritems')
-
-    def test_safe_to_remove_while_keys(self):
-        self.assertSafeIter('keys')
-
-    def test_safe_to_remove_while_itervalues(self):
-        self.assertSafeIter('itervalues')
-
-    def test_items(self):
-        c = LRUCache()
-        c.update(a=1, b=2, c=3)
-        self.assertTrue(list(items(c)))
-
-
 class test_AttributeDict(Case):
 
     def test_getattr__setattr(self):

celery/tests/utilities/test_functional.py

@@ -0,0 +1,103 @@
+from __future__ import absolute_import
+
+import pickle
+
+from celery.five import THREAD_TIMEOUT_MAX, items, range
+from celery.utils.functional import LRUCache
+
+from celery.tests.utils import Case
+
+
+class test_LRUCache(Case):
+
+    def test_expires(self):
+        limit = 100
+        x = LRUCache(limit=limit)
+        slots = list(range(limit * 2))
+        for i in slots:
+            x[i] = i
+        self.assertListEqual(list(x.keys()), list(slots[limit:]))
+        self.assertTrue(x.items())
+        self.assertTrue(x.values())
+
+    def test_is_pickleable(self):
+        x = LRUCache(limit=10)
+        x.update(luke=1, leia=2)
+        y = pickle.loads(pickle.dumps(x))
+        self.assertEqual(y.limit, y.limit)
+        self.assertEqual(y, x)
+
+    def test_update_expires(self):
+        limit = 100
+        x = LRUCache(limit=limit)
+        slots = list(range(limit * 2))
+        for i in slots:
+            x.update({i: i})
+
+        self.assertListEqual(list(x.keys()), list(slots[limit:]))
+
+    def test_least_recently_used(self):
+        x = LRUCache(3)
+
+        x[1], x[2], x[3] = 1, 2, 3
+        self.assertEqual(list(x.keys()), [1, 2, 3])
+
+        x[4], x[5] = 4, 5
+        self.assertEqual(list(x.keys()), [3, 4, 5])
+
+        # access 3, which makes it the last used key.
+        x[3]
+        x[6] = 6
+        self.assertEqual(list(x.keys()), [5, 3, 6])
+
+        x[7] = 7
+        self.assertEqual(list(x.keys()), [3, 6, 7])
+
+    def assertSafeIter(self, method, interval=0.01, size=10000):
+        from threading import Thread, Event
+        from time import sleep
+        x = LRUCache(size)
+        x.update(zip(range(size), range(size)))
+
+        class Burglar(Thread):
+
+            def __init__(self, cache):
+                self.cache = cache
+                self._is_shutdown = Event()
+                self._is_stopped = Event()
+                Thread.__init__(self)
+
+            def run(self):
+                while not self._is_shutdown.isSet():
+                    try:
+                        self.cache.data.popitem(last=False)
+                    except KeyError:
+                        break
+                self._is_stopped.set()
+
+            def stop(self):
+                self._is_shutdown.set()
+                self._is_stopped.wait()
+                self.join(THREAD_TIMEOUT_MAX)
+
+        burglar = Burglar(x)
+        burglar.start()
+        try:
+            for _ in getattr(x, method)():
+                sleep(0.0001)
+        finally:
+            burglar.stop()
+
+    def test_safe_to_remove_while_iteritems(self):
+        self.assertSafeIter('iteritems')
+
+    def test_safe_to_remove_while_keys(self):
+        self.assertSafeIter('keys')
+
+    def test_safe_to_remove_while_itervalues(self):
+        self.assertSafeIter('itervalues')
+
+    def test_items(self):
+        c = LRUCache()
+        c.update(a=1, b=2, c=3)
+        self.assertTrue(list(items(c)))

celery/utils/functional.py

@@ -8,6 +8,7 @@
 """
 from __future__ import absolute_import
 
+import sys
 import threading
 
 from functools import wraps
@@ -42,16 +43,6 @@ class LRUCache(UserDict):
             value = self[key] = self.data.pop(key)
         return value
 
-    def keys(self):
-        # userdict.keys in py3k calls __getitem__
-        return keys(self.data)
-
-    def values(self):
-        return list(self._iterate_values())
-
-    def items(self):
-        return list(self._iterate_items())
-
     def update(self, *args, **kwargs):
         with self.mutex:
             data, limit = self.data, self.limit
@@ -88,6 +79,11 @@ class LRUCache(UserDict):
                 pass
     itervalues = _iterate_values
 
+    def _iterate_keys(self):
+        # userdict.keys in py3k calls __getitem__
+        return keys(self.data)
+    iterkeys = _iterate_keys
+
     def incr(self, key, delta=1):
         with self.mutex:
             # this acts as memcached does- store as a string, but return a
@@ -105,6 +101,21 @@ class LRUCache(UserDict):
         self.__dict__ = state
         self.mutex = threading.RLock()
 
+    if sys.version_info[0] == 3:  # pragma: no cover
+        keys = _iterate_keys
+        values = _iterate_values
+        items = _iterate_items
+    else:  # noqa
+
+        def keys(self):
+            return list(self._iterate_keys())
+
+        def values(self):
+            return list(self._iterate_values())
+
+        def items(self):
+            return list(self._iterate_items())
+
 
 def is_list(l, scalars=(dict, string_t)):
     """Returns true if object is list-like, but not a dict or string."""

celery/utils/imports.py

@@ -24,7 +24,7 @@ class NotAPackage(Exception):
     pass
 
 
-if sys.version_info > (3, 3):
+if sys.version_info > (3, 3):  # pragma: no cover
     def qualname(obj):
         if not hasattr(obj, '__name__') and hasattr(obj, '__class__'):
             obj = obj.__class__