100% coverage for celery.apps.beat

celery/tests/test_bin/test_celerybeat.py

@@ -1,6 +1,7 @@
 import logging
 import sys
-from celery.tests.utils import unittest
+
+from kombu.tests.utils import redirect_stdouts
 
 from celery import beat
 from celery import platforms
@@ -9,6 +10,7 @@ from celery.bin import celerybeat as celerybeat_bin
 from celery.apps import beat as beatapp
 from celery.utils.compat import defaultdict
 
+from celery.tests.utils import unittest
 
 class MockedShelveModule(object):
     shelves = defaultdict(lambda: {})
@@ -44,6 +46,13 @@ class MockBeat2(beatapp.Beat):
         pass
 
 
+class MockBeat3(beatapp.Beat):
+    Service = MockService
+
+    def install_sync_handler(self, b):
+        raise TypeError("xxx")
+
+
 class test_Beat(unittest.TestCase):
 
     def test_loglevel_string(self):
@@ -91,6 +100,60 @@ class test_Beat(unittest.TestCase):
         self.assertTrue(MockService.in_sync)
         MockService.in_sync = False
 
+    def test_setup_logging(self):
+        import sys
+        b = beatapp.Beat()
+        b.redirect_stdouts = False
+        b.setup_logging()
+        self.assertRaises(AttributeError, getattr, sys.stdout, "logger")
+
+    @redirect_stdouts
+    def test_logs_errors(self, stdout, stderr):
+        class MockLogger(object):
+            _critical = []
+
+            def debug(self, *args, **kwargs):
+                pass
+
+            def critical(self, msg, *args, **kwargs):
+                self._critical.append(msg)
+
+        logger = MockLogger()
+        b = MockBeat3(socket_timeout=None)
+        b.start_scheduler(logger)
+
+        self.assertTrue(logger._critical)
+
+    @redirect_stdouts
+    def test_use_pidfile(self, stdout, stderr):
+        from celery import platforms
+
+
+        class create_pidlock(object):
+            instance = [None]
+
+            def __init__(self, file):
+                self.file = file
+                self.instance[0] = self
+
+            def acquire(self):
+                self.acquired = True
+
+                class Object(object):
+                    def release(self):
+                        pass
+
+                return Object()
+
+        prev, platforms.create_pidlock = platforms.create_pidlock, \
+                                         create_pidlock
+        try:
+            b = MockBeat2(pidfile="pidfilelockfilepid", socket_timeout=None)
+            b.start_scheduler()
+            self.assertTrue(create_pidlock.instance[0].acquired)
+        finally:
+            platforms.create_pidlock = prev
+
 
 class test_div(unittest.TestCase):