100% coverage for celery.utils

celery/tests/utilities/test_functional.py

@@ -2,8 +2,17 @@ from __future__ import absolute_import
 
 import pickle
 
-from celery.five import THREAD_TIMEOUT_MAX, items, range
-from celery.utils.functional import LRUCache
+from kombu.utils.functional import promise
+
+from celery.five import THREAD_TIMEOUT_MAX, items, range, nextfun
+from celery.utils.functional import (
+    LRUCache,
+    firstmethod,
+    first,
+    mpromise,
+    padlist,
+    maybe_list,
+)
 
 from celery.tests.utils import Case
 
@@ -101,3 +110,71 @@ class test_LRUCache(Case):
         c = LRUCache()
         c.update(a=1, b=2, c=3)
         self.assertTrue(list(items(c)))
+
+
+class test_utils(Case):
+
+    def test_padlist(self):
+        self.assertListEqual(
+            padlist(['George', 'Costanza', 'NYC'], 3),
+            ['George', 'Costanza', 'NYC'],
+        )
+        self.assertListEqual(
+            padlist(['George', 'Costanza'], 3),
+            ['George', 'Costanza', None],
+        )
+        self.assertListEqual(
+            padlist(['George', 'Costanza', 'NYC'], 4, default='Earth'),
+            ['George', 'Costanza', 'NYC', 'Earth'],
+        )
+
+    def test_firstmethod_AttributeError(self):
+        self.assertIsNone(firstmethod('foo')([object()]))
+
+    def test_firstmethod_promises(self):
+
+        class A(object):
+
+            def __init__(self, value=None):
+                self.value = value
+
+            def m(self):
+                return self.value
+
+        self.assertEqual('four', firstmethod('m')([
+            A(), A(), A(), A('four'), A('five')]))
+        self.assertEqual('four', firstmethod('m')([
+            A(), A(), A(), promise(lambda: A('four')), A('five')]))
+
+    def test_first(self):
+        iterations = [0]
+
+        def predicate(value):
+            iterations[0] += 1
+            if value == 5:
+                return True
+            return False
+
+        self.assertEqual(5, first(predicate, range(10)))
+        self.assertEqual(iterations[0], 6)
+
+        iterations[0] = 0
+        self.assertIsNone(first(predicate, range(10, 20)))
+        self.assertEqual(iterations[0], 10)
+
+    def test_maybe_list(self):
+        self.assertEqual(maybe_list(1), [1])
+        self.assertEqual(maybe_list([1]), [1])
+        self.assertIsNone(maybe_list(None))
+
+
+class test_mpromise(Case):
+
+    def test_is_memoized(self):
+
+        it = iter(range(20, 30))
+        p = mpromise(nextfun(it))
+        self.assertEqual(p(), 20)
+        self.assertTrue(p.evaluated)
+        self.assertEqual(p(), 20)
+        self.assertEqual(repr(p), '20')

celery/tests/utilities/test_platforms.py

@@ -123,6 +123,7 @@ if not platforms.IS_WINDOWS:
 
             class pw_struct(object):
                 pw_gid = 50001
+
             def raise_on_second_call(*args, **kwargs):
                 setuid.side_effect = OSError()
                 setuid.side_effect.errno = errno.EPERM

celery/tests/utilities/test_serialization.py

@@ -10,7 +10,6 @@ from celery.utils.serialization import (
 from celery.tests.utils import Case, mask_modules
 
 
-
 class test_AAPickle(Case):
 
     def test_no_cpickle(self):

celery/tests/utilities/test_sysinfo.py

@@ -2,7 +2,7 @@ from __future__ import absolute_import
 
 import os
 
-from mock import Mock, patch
+from mock import patch
 from nose import SkipTest
 
 from celery.utils.sysinfo import load_average, df
@@ -26,7 +26,7 @@ class test_df(Case):
 
     def test_df(self):
         try:
-            from posix import statvfs_result
+            from posix import statvfs_result  # noqa
         except ImportError:
             raise SkipTest('statvfs not available')
         x = df('/')

celery/tests/utilities/test_info.py → celery/tests/utilities/test_text.py

@@ -1,7 +1,14 @@
 from __future__ import absolute_import
 
 from celery import Celery
-from celery.utils.text import indent, ensure_2lines
+from celery.utils.text import (
+    indent,
+    ensure_2lines,
+    abbr,
+    truncate,
+    abbrtask,
+    pretty,
+)
 from celery.tests.utils import Case
 
 RANDTEXT = """\
@@ -54,3 +61,30 @@ class test_Info(Case):
         self.assertEqual(
             len(ensure_2lines('foo\nbar').splitlines()), 2,
         )
+
+
+class test_utils(Case):
+
+    def test_truncate_text(self):
+        self.assertEqual(truncate('ABCDEFGHI', 3), 'ABC...')
+        self.assertEqual(truncate('ABCDEFGHI', 10), 'ABCDEFGHI')
+
+    def test_abbr(self):
+        self.assertEqual(abbr(None, 3), '???')
+        self.assertEqual(abbr('ABCDEFGHI', 6), 'ABC...')
+        self.assertEqual(abbr('ABCDEFGHI', 20), 'ABCDEFGHI')
+        self.assertEqual(abbr('ABCDEFGHI', 6, None), 'ABCDEF')
+
+    def test_abbrtask(self):
+        self.assertEqual(abbrtask(None, 3), '???')
+        self.assertEqual(
+            abbrtask('feeds.tasks.refresh', 10),
+            '[.]refresh',
+        )
+        self.assertEqual(
+            abbrtask('feeds.tasks.refresh', 30),
+            'feeds.tasks.refresh',
+        )
+
+    def test_pretty(self):
+        self.assertTrue(pretty(('a', 'b', 'c')))

celery/tests/utilities/test_threads.py

@@ -1,6 +1,6 @@
 from __future__ import absolute_import
 
-from mock import Mock, patch
+from mock import patch
 
 from celery.utils.threads import (
     _LocalStack,
@@ -30,6 +30,11 @@ class test_bgThread(Case):
                     t.run()
                 _exit.assert_called_with(1)
 
+    def test_interface(self):
+        x = bgThread()
+        with self.assertRaises(NotImplementedError):
+            x.body()
+
 
 class test_Local(Case):
 
@@ -102,5 +107,3 @@ class test_LocalManager(Case):
             release.assert_called_with(loc)
 
         self.assertTrue(repr(x))
-
-

celery/tests/utilities/test_timer2.py

@@ -30,7 +30,7 @@ class test_Entry(Case):
         self.assertTrue(tref.cancelled)
 
     def test_repr(self):
-        tref = timer2.Entry(lambda x: x (1, ), {})
+        tref = timer2.Entry(lambda x: x(1, ), {})
         self.assertTrue(repr(tref))
 
 
@@ -46,7 +46,6 @@ class test_Schedule(Case):
 
         self.assertIs(x.schedule, x)
 
-
     def test_handle_error(self):
         from datetime import datetime
         to_timestamp = timer2.to_timestamp

celery/tests/utilities/test_timeutils.py

@@ -1,7 +1,6 @@
 from __future__ import absolute_import
 
 import pytz
-import time
 
 from datetime import datetime, timedelta, tzinfo
 from mock import Mock, patch
@@ -55,7 +54,6 @@ class test_LocalTimezone(Case):
             self.assertTrue(y.tzname(datetime.now()))
 
 
-
 class test_iso8601(Case):
 
     def test_parse_with_timezone(self):
@@ -166,8 +164,10 @@ class test_make_aware(Case):
         self.assertEqual(wtz.tzinfo, tz)
 
     def test_when_has_localize(self):
+
         class tzz(tzinfo):
             raises = False
+
             def localize(self, dt, is_dst=None):
                 self.localized = True
                 if self.raises and is_dst is None:
@@ -175,12 +175,12 @@ class test_make_aware(Case):
                     raise AmbiguousTimeError()
 
         tz = tzz()
-        wtz = make_aware(datetime.utcnow(), tz)
+        make_aware(datetime.utcnow(), tz)
         self.assertTrue(tz.localized)
 
         tz2 = tzz()
         tz2.raises = True
-        wtz = make_aware(datetime.utcnow(), tz2)
+        make_aware(datetime.utcnow(), tz2)
         self.assertTrue(tz2.localized)
         self.assertTrue(tz2.raised)
 
@@ -199,8 +199,10 @@ class test_localize(Case):
         self.assertTrue(localize(make_aware(datetime.utcnow(), tz), tz))
 
     def test_when_has_nornalize(self):
+
         class tzz(tzinfo):
             raises = None
+
             def normalize(self, dt, **kwargs):
                 self.normalized = True
                 if self.raises and kwargs and kwargs.get('is_dst') is None:
@@ -208,18 +210,18 @@ class test_localize(Case):
                     raise self.raises
 
         tz = tzz()
-        wtz = localize(make_aware(datetime.utcnow(), tz), tz)
+        localize(make_aware(datetime.utcnow(), tz), tz)
         self.assertTrue(tz.normalized)
 
         tz2 = tzz()
         tz2.raises = AmbiguousTimeError()
-        wtz = localize(make_aware(datetime.utcnow(), tz2), tz2)
+        localize(make_aware(datetime.utcnow(), tz2), tz2)
         self.assertTrue(tz2.normalized)
         self.assertTrue(tz2.raised)
 
         tz3 = tzz()
         tz3.raises = TypeError()
-        wtz = localize(make_aware(datetime.utcnow(), tz3), tz3)
+        localize(make_aware(datetime.utcnow(), tz3), tz3)
         self.assertTrue(tz3.normalized)
         self.assertTrue(tz3.raised)
 

celery/tests/utilities/test_utils.py

@@ -1,15 +1,21 @@
 from __future__ import absolute_import
 
-from kombu.utils.functional import promise
-
-from mock import patch
-
-from celery import utils
-from celery.five import nextfun, range
-from celery.utils import text
-from celery.utils import functional
-from celery.utils.functional import mpromise, maybe_list
-from celery.utils.threads import bgThread
+import pytz
+
+from datetime import datetime, date, time, timedelta
+from mock import Mock, patch
+
+from kombu import Queue
+
+from celery.utils import (
+    chunks,
+    is_iterable,
+    cached_property,
+    warn_deprecated,
+    worker_direct,
+    gen_task_name,
+    jsonify,
+)
 from celery.tests.utils import Case
 
 
@@ -17,12 +23,40 @@ def double(x):
     return x * 2
 
 
-class test_bgThread_interface(Case):
+class test_worker_direct(Case):
+
+    def test_returns_if_queue(self):
+        q = Queue('foo')
+        self.assertIs(worker_direct(q), q)
+
+
+class test_gen_task_name(Case):
 
-    def test_body(self):
-        x = bgThread()
-        with self.assertRaises(NotImplementedError):
-            x.body()
+    def test_no_module(self):
+        app = Mock()
+        app.name == '__main__'
+        self.assertTrue(gen_task_name(app, 'foo', 'axsadaewe'))
+
+
+class test_jsonify(Case):
+
+    def test_simple(self):
+        self.assertTrue(jsonify(Queue('foo')))
+        self.assertTrue(jsonify(['foo', 'bar', 'baz']))
+        self.assertTrue(jsonify({'foo': 'bar'}))
+        self.assertTrue(jsonify(datetime.utcnow()))
+        self.assertTrue(jsonify(datetime.utcnow().replace(tzinfo=pytz.utc)))
+        self.assertTrue(jsonify(datetime.utcnow().replace(microsecond=0)))
+        self.assertTrue(jsonify(date(2012, 1, 1)))
+        self.assertTrue(jsonify(time(hour=1, minute=30)))
+        self.assertTrue(jsonify(time(hour=1, minute=30, microsecond=3)))
+        self.assertTrue(jsonify(timedelta(seconds=30)))
+        self.assertTrue(jsonify(10))
+        self.assertTrue(jsonify(10.3))
+        self.assertTrue(jsonify('hello'))
+
+        with self.assertRaises(ValueError):
+            jsonify(object())
 
 
 class test_chunks(Case):
@@ -30,21 +64,21 @@ class test_chunks(Case):
     def test_chunks(self):
 
         # n == 2
-        x = utils.chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 2)
+        x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 2)
         self.assertListEqual(
             list(x),
             [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]],
         )
 
         # n == 3
-        x = utils.chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 3)
+        x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 3)
         self.assertListEqual(
             list(x),
             [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]],
         )
 
         # n == 2 (exact)
-        x = utils.chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]), 2)
+        x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]), 2)
         self.assertListEqual(
             list(x),
             [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]],
@@ -55,111 +89,21 @@ class test_utils(Case):
 
     def test_is_iterable(self):
         for a in 'f', ['f'], ('f', ), {'f': 'f'}:
-            self.assertTrue(utils.is_iterable(a))
+            self.assertTrue(is_iterable(a))
         for b in object(), 1:
-            self.assertFalse(utils.is_iterable(b))
-
-    def test_padlist(self):
-        self.assertListEqual(
-            functional.padlist(['George', 'Costanza', 'NYC'], 3),
-            ['George', 'Costanza', 'NYC'],
-        )
-        self.assertListEqual(
-            functional.padlist(['George', 'Costanza'], 3),
-            ['George', 'Costanza', None],
-        )
-        self.assertListEqual(
-            functional.padlist(['George', 'Costanza', 'NYC'], 4,
-                               default='Earth'),
-            ['George', 'Costanza', 'NYC', 'Earth'],
-        )
-
-    def test_firstmethod_AttributeError(self):
-        self.assertIsNone(functional.firstmethod('foo')([object()]))
-
-    def test_firstmethod_promises(self):
-
-        class A(object):
-
-            def __init__(self, value=None):
-                self.value = value
-
-            def m(self):
-                return self.value
-
-        self.assertEqual('four', functional.firstmethod('m')([
-            A(), A(), A(), A('four'), A('five')]))
-        self.assertEqual('four', functional.firstmethod('m')([
-            A(), A(), A(), promise(lambda: A('four')), A('five')]))
-
-    def test_first(self):
-        iterations = [0]
-
-        def predicate(value):
-            iterations[0] += 1
-            if value == 5:
-                return True
-            return False
-
-        self.assertEqual(5, functional.first(predicate, range(10)))
-        self.assertEqual(iterations[0], 6)
-
-        iterations[0] = 0
-        self.assertIsNone(functional.first(predicate, range(10, 20)))
-        self.assertEqual(iterations[0], 10)
-
-    def test_truncate_text(self):
-        self.assertEqual(text.truncate('ABCDEFGHI', 3), 'ABC...')
-        self.assertEqual(text.truncate('ABCDEFGHI', 10), 'ABCDEFGHI')
-
-    def test_abbr(self):
-        self.assertEqual(text.abbr(None, 3), '???')
-        self.assertEqual(text.abbr('ABCDEFGHI', 6), 'ABC...')
-        self.assertEqual(text.abbr('ABCDEFGHI', 20), 'ABCDEFGHI')
-        self.assertEqual(text.abbr('ABCDEFGHI', 6, None), 'ABCDEF')
-
-    def test_abbrtask(self):
-        self.assertEqual(text.abbrtask(None, 3), '???')
-        self.assertEqual(
-            text.abbrtask('feeds.tasks.refresh', 10),
-            '[.]refresh',
-        )
-        self.assertEqual(
-            text.abbrtask('feeds.tasks.refresh', 30),
-            'feeds.tasks.refresh',
-        )
-
-    def test_pretty(self):
-        self.assertTrue(text.pretty(('a', 'b', 'c')))
+            self.assertFalse(is_iterable(b))
 
     def test_cached_property(self):
 
         def fun(obj):
             return fun.value
 
-        x = utils.cached_property(fun)
+        x = cached_property(fun)
         self.assertIs(x.__get__(None), x)
         self.assertIs(x.__set__(None, None), x)
         self.assertIs(x.__delete__(None), x)
 
-    def test_maybe_list(self):
-        self.assertEqual(maybe_list(1), [1])
-        self.assertEqual(maybe_list([1]), [1])
-        self.assertIsNone(maybe_list(None))
-
     @patch('warnings.warn')
     def test_warn_deprecated(self, warn):
-        utils.warn_deprecated('Foo')
+        warn_deprecated('Foo')
         self.assertTrue(warn.called)
-
-
-class test_mpromise(Case):
-
-    def test_is_memoized(self):
-
-        it = iter(range(20, 30))
-        p = mpromise(nextfun(it))
-        self.assertEqual(p(), 20)
-        self.assertTrue(p.evaluated)
-        self.assertEqual(p(), 20)
-        self.assertEqual(repr(p), '20')