98% coverage for celery.local

celery/local.py

@@ -34,88 +34,11 @@ def _default_cls_attr(name, type_, cls_value):
     def __get__(self, obj, cls=None):
         return self.__getter(obj) if obj is not None else self
 
-    def __set__(self, obj, value):
-        raise AttributeError('readonly attribute')
-
     return type(name, (type_, ), {
-        '__new__': __new__, '__get__': __get__, '__set__': __set__,
+        '__new__': __new__, '__get__': __get__,
     })
 
 
-class _cls_spec(str):
-
-    def __new__(cls, getter):
-        s = str.__new__(cls, getter.__module__)
-        s.__getter = getter
-        return s
-
-    def __get__(self, obj, cls=None):
-        if obj is not None:
-            return self.__getter(obj)
-        return self
-
-    def __set__(self, obj, value):
-        raise AttributeError('cannot set attribute')
-
-
-def symbol_by_name(name, aliases={}, imp=None, package=None,
-                   sep='.', default=None, **kwargs):
-    """Get symbol by qualified name.
-
-    The name should be the full dot-separated path to the class::
-
-        modulename.ClassName
-
-    Example::
-
-        celery.concurrency.processes.TaskPool
-                                    ^- class name
-
-    or using ':' to separate module and symbol::
-
-        celery.concurrency.processes:TaskPool
-
-    If `aliases` is provided, a dict containing short name/long name
-    mappings, the name is looked up in the aliases first.
-
-    Examples:
-
-        >>> symbol_by_name('celery.concurrency.processes.TaskPool')
-        <class 'celery.concurrency.processes.TaskPool'>
-
-        >>> symbol_by_name('default', {
-        ...     'default': 'celery.concurrency.processes.TaskPool'})
-        <class 'celery.concurrency.processes.TaskPool'>
-
-        # Does not try to look up non-string names.
-        >>> from celery.concurrency.processes import TaskPool
-        >>> symbol_by_name(TaskPool) is TaskPool
-        True
-
-    """
-    if imp is None:
-        imp = importlib.import_module
-
-    if not isinstance(name, string_t):
-        return name                                 # already a class
-
-    name = aliases.get(name) or name
-    sep = ':' if ':' in name else sep
-    module_name, _, cls_name = name.rpartition(sep)
-    if not module_name:
-        cls_name, module_name = None, package if package else cls_name
-    try:
-        try:
-            module = imp(module_name, package=package, **kwargs)
-        except ValueError as exc:
-            raise ValueError("Couldn't import %r: %s" % (name, exc))
-        return getattr(module, cls_name) if cls_name else module
-    except (ImportError, AttributeError):
-        if default is None:
-            raise
-    return default
-
-
 def try_import(module, default=None):
     """Try to import and return module, or return
     None if the module does not exist."""
@@ -265,7 +188,7 @@ class Proxy(object):
     __oct__ = lambda x: oct(x._get_current_object())
     __hex__ = lambda x: hex(x._get_current_object())
     __index__ = lambda x: x._get_current_object().__index__()
-    __coerce__ = lambda x, o: x.__coerce__(x, o)
+    __coerce__ = lambda x, o: x._get_current_object().__coerce__(o)
     __enter__ = lambda x: x._get_current_object().__enter__()
     __exit__ = lambda x, *a, **kw: x._get_current_object().__exit__(*a, **kw)
     __reduce__ = lambda x: x._get_current_object().__reduce__()
@@ -307,7 +230,8 @@ class PromiseProxy(Proxy):
             for attr in _clean:
                 try:
                     object.__delattr__(self, attr)
-                except AttributeError:  # May mask errors so ignore
+                except AttributeError:  # pragma: no cover
+                    # May mask errors so ignore
                     pass
 
 

celery/tests/utilities/test_local.py

@@ -1,7 +1,14 @@
 from __future__ import absolute_import, unicode_literals
 
+from mock import Mock
+
 from celery.five import string, long_t
-from celery.local import Proxy, PromiseProxy, maybe_evaluate, try_import
+from celery.local import (
+    Proxy,
+    PromiseProxy,
+    maybe_evaluate,
+    try_import,
+)
 from celery.tests.utils import Case
 
 
@@ -39,6 +46,12 @@ class test_Proxy(Case):
         self.assertEqual(x.__class__, type(real))
         self.assertEqual(x.__dict__, real.__dict__)
         self.assertEqual(repr(x), repr(real))
+        self.assertTrue(x.__module__)
+
+    def test_get_current_local(self):
+        x = Proxy(lambda: 10)
+        object.__setattr__(x, '_Proxy_local', Mock())
+        self.assertTrue(x._get_current_object())
 
     def test_bool(self):
 
@@ -144,6 +157,40 @@ class test_Proxy(Case):
         self.assertIn(10, x)
         self.assertEqual(len(x), 3)
         self.assertTrue(iter(x))
+        x[0:2] = [1, 2]
+        del(x[0:2])
+        self.assertTrue(str(x))
+        self.assertEqual(x.__cmp__(object()), -1)
+
+    def test_complex_cast(self):
+
+        class O(object):
+
+            def __complex__(self):
+                return 10.333
+
+        o = Proxy(O)
+        self.assertEqual(o.__complex__(), 10.333)
+
+    def test_index(self):
+
+        class O(object):
+
+            def __index__(self):
+                return 1
+
+        o = Proxy(O)
+        self.assertEqual(o.__index__(), 1)
+
+    def test_coerce(self):
+
+        class O(object):
+
+            def __coerce__(self, other):
+                return self, other
+
+        o = Proxy(O)
+        self.assertTrue(o.__coerce__(3))
 
     def test_int(self):
         self.assertEqual(Proxy(lambda: 10) + 1, Proxy(lambda: 11))
@@ -166,6 +213,9 @@ class test_Proxy(Case):
         self.assertTrue(Proxy(lambda: 10) <= Proxy(lambda: 10))
         self.assertTrue(Proxy(lambda: 10) == Proxy(lambda: 10))
         self.assertTrue(Proxy(lambda: 20) != Proxy(lambda: 10))
+        self.assertTrue(Proxy(lambda: 100).__divmod__(30))
+        self.assertTrue(Proxy(lambda: 100).__truediv__(30))
+        self.assertTrue(abs(Proxy(lambda: -100)))
 
         x = Proxy(lambda: 10)
         x -= 1
@@ -277,7 +327,9 @@ class test_PromiseProxy(Case):
 
     def test_maybe_evaluate(self):
         x = PromiseProxy(lambda: 30)
+        self.assertFalse(x.__evaluated__())
         self.assertEqual(maybe_evaluate(x), 30)
         self.assertEqual(maybe_evaluate(x), 30)
 
         self.assertEqual(maybe_evaluate(30), 30)
+        self.assertTrue(x.__evaluated__())