Browse Source

more tests

Ask Solem 12 years ago
parent
commit
edbb287ddb

+ 11 - 0
celery/tests/tasks/test_result.py

@@ -78,6 +78,17 @@ class test_AsyncResult(AppCase):
         self.assertTrue(x.children)
         self.assertEqual(len(x.children), 3)
 
+    def test_get_children(self):
+        tid = uuid()
+        x = AsyncResult(tid)
+        child = [AsyncResult(uuid()).serializable() for i in xrange(10)]
+        x.backend._cache[tid] = {"children": child}
+        self.assertTrue(x.children)
+        self.assertEqual(len(x.children), 10)
+
+        x.backend._cache[tid] = {"result": None}
+        self.assertIsNone(x.children)
+
     def test_build_graph_get_leaf_collect(self):
         x = AsyncResult("1")
         x.backend._cache["1"] = {"status": states.SUCCESS, "result": None}

+ 30 - 0
celery/tests/utilities/test_timer2.py

@@ -33,6 +33,14 @@ class test_Entry(Case):
 
 class test_Schedule(Case):
 
+    def test_supports_Timer_interface(self):
+        x = timer2.Schedule()
+        x.stop()
+
+        tref = Mock()
+        x.cancel(tref)
+        tref.cancel.assert_called_with()
+
     def test_handle_error(self):
         from datetime import datetime
         mktime = timer2.mktime
@@ -150,3 +158,25 @@ class test_Timer(Case):
         t._is_shutdown.set()
         t.run()
         t._is_stopped.set.assert_called_with()
+
+    def test_to_timestamp(self):
+        self.assertIs(timer2.to_timestamp(3.13), 3.13)
+
+    def test_test_enter(self):
+        t = timer2.Timer()
+        t._do_enter = Mock()
+        e = Mock()
+        t.enter(e, 13, 0)
+        t._do_enter.assert_called_with("enter", e, 13, priority=0)
+
+    def test_test_enter_after(self):
+        t = timer2.Timer()
+        t._do_enter = Mock()
+        t.enter_after()
+        t._do_enter.assert_called_with("enter_after")
+
+    def test_cancel(self):
+        t = timer2.Timer()
+        tref = Mock()
+        t.cancel(tref)
+        tref.cancel.assert_called_with()

+ 2 - 2
celery/utils/compat.py

@@ -157,10 +157,10 @@ else:
 
 ############## format(int, ',d') ##########################
 
-if sys.version_info >= (2, 7):
+if sys.version_info >= (2, 7):  # pragma: no cover
     def format_d(i):
         return format(i, ',d')
-else:
+else:  # pragma: no cover
     def format_d(i):  # noqa
         s = '%d' % i
         groups = []

+ 2 - 2
celery/utils/timer2.py

@@ -57,7 +57,7 @@ class Entry(object):
         return "<TimerEntry: %s(*%r, **%r)" % (
                 self.fun.__name__, self.args, self.kwargs)
 
-    if sys.version_info >= (3, 0):
+    if sys.version_info[0] == 3:  # pragma: no cover
 
         def __hash__(self):
             return hash("|".join(map(repr, (self.fun, self.args,
@@ -217,7 +217,7 @@ class Timer(Thread):
     on_tick = None
     _timer_count = count(1).next
 
-    if TIMER_DEBUG:
+    if TIMER_DEBUG:  # pragma: no cover
         def start(self, *args, **kwargs):
             import traceback
             print("- Timer starting")

+ 1 - 1
setup.cfg

@@ -3,7 +3,7 @@ where = celery/tests
 cover3-branch = 1
 cover3-html = 1
 cover3-package = celery
-cover3-exclude = celery.tests.*
+cover3-exclude = celery.utils.debug,celery.tests.*
 
 [build_sphinx]
 source-dir = docs/