100% Coverage for registry routes states schedules messaging

celery/tests/test_messaging.py

@@ -0,0 +1,17 @@
+from celery import messaging
+from celery.tests.utils import unittest
+
+
+class test_compat_messaging_module(unittest.TestCase):
+
+    def test_with_connection(self):
+
+        def foo(**kwargs):
+            pass
+
+        self.assertTrue(messaging.with_connection(foo))
+
+    def test_get_consume_set(self):
+        conn = messaging.establish_connection()
+        messaging.get_consumer_set(conn).close()
+        conn.close()

celery/tests/test_registry.py

@@ -39,6 +39,11 @@ class TestTaskRegistry(unittest.TestCase):
         self.assertRegisterUnregisterCls(r, TestTask)
         self.assertRegisterUnregisterCls(r, TestPeriodicTask)
 
+        r.register(TestPeriodicTask)
+        r.unregister(TestPeriodicTask.name)
+        self.assertNotIn(TestPeriodicTask, r)
+        r.register(TestPeriodicTask)
+
         tasks = dict(r)
         self.assertIsInstance(tasks.get(TestTask.name), TestTask)
         self.assertIsInstance(tasks.get(TestPeriodicTask.name),

celery/tests/test_routes.py

@@ -95,6 +95,12 @@ class test_lookup_route(unittest.TestCase):
                                        route)
         self.assertNotIn("queue", route)
 
+    @with_queues(foo=a_queue, bar=b_queue)
+    def test_expand_destaintion_string(self):
+        x = routes.Router({}, app_or_default().conf.CELERY_QUEUES)
+        dest = x.expand_destination("foo")
+        self.assertEqual(dest["exchange"], "fooexchange")
+
     @with_queues(foo=a_queue, bar=b_queue)
     def test_lookup_paths_traversed(self):
         R = routes.prepare(({"celery.xaza": {"queue": "bar"}},

celery/tests/test_states.py

@@ -26,3 +26,6 @@ class test_state_precedence(unittest.TestCase):
         self.assertLess(state("CRASHED"), state(states.SUCCESS))
         self.assertLess(state("CRASHED"), state(states.FAILURE))
         self.assertTrue(state(states.REVOKED) < state("CRASHED"))
+        self.assertTrue(state(states.REVOKED) <= state("CRASHED"))
+        self.assertTrue(state("CRASHED") >= state(states.REVOKED))
+

celery/tests/test_task.py

@@ -592,6 +592,25 @@ class test_crontab_parser(unittest.TestCase):
     def test_parse_errors_on_negative_number(self):
         self.assertRaises(ParseException, crontab_parser(60).parse, '-20')
 
+    def test_expand_cronspec_eats_iterables(self):
+        self.assertEqual(crontab._expand_cronspec(iter([1, 2, 3]), 100),
+                         set([1, 2, 3]))
+
+    def test_expand_cronspec_invalid_type(self):
+        self.assertRaises(TypeError, crontab._expand_cronspec, object(), 100)
+
+    def test_repr(self):
+        self.assertIn("*", repr(crontab("*")))
+
+    def test_eq(self):
+        self.assertEqual(crontab(day_of_week="1, 2"),
+                         crontab(day_of_week="1-2"))
+        self.assertEqual(crontab(minute="1", hour="2", day_of_week="5"),
+                         crontab(minute="1", hour="2", day_of_week="5"))
+        self.assertNotEqual(crontab(minute="1"), crontab(minute="2"))
+        self.assertFalse(object() == crontab(minute="1"))
+        self.assertFalse(crontab(minute="1") == object())
+
 
 class test_crontab_remaining_estimate(unittest.TestCase):