100% coverage for celery.apps.*

celery/apps/worker.py

@@ -270,11 +270,11 @@ def _shutdown_handler(worker, sig='TERM', how='Warm',
 install_worker_term_handler = partial(
     _shutdown_handler, sig='SIGTERM', how='Warm', exc=SystemExit,
 )
-if not is_jython:
+if not is_jython:  # pragma: no cover
     install_worker_term_hard_handler = partial(
         _shutdown_handler, sig='SIGQUIT', how='Cold', exc=SystemTerminate,
     )
-else:
+else:  # pragma: no cover
     install_worker_term_handler = \
         install_worker_term_hard_handler = lambda *a, **kw: None
 
@@ -282,11 +282,11 @@ else:
 def on_SIGINT(worker):
     safe_say('worker: Hitting Ctrl+C again will terminate all running tasks!')
     install_worker_term_hard_handler(worker, sig='SIGINT')
-if not is_jython:
+if not is_jython:  # pragma: no cover
     install_worker_int_handler = partial(
         _shutdown_handler, sig='SIGINT', callback=on_SIGINT
     )
-else:
+else:  # pragma: no cover
     install_worker_int_handler = lambda *a, **kw: None
 
 

celery/tests/bin/test_beat.py

@@ -14,7 +14,7 @@ from celery.app import app_or_default
 from celery.bin import beat as beat_bin
 from celery.apps import beat as beatapp
 
-from celery.tests.utils import AppCase
+from celery.tests.utils import AppCase, Mock
 
 
 class MockedShelveModule(object):
@@ -70,6 +70,16 @@ class test_Beat(AppCase):
         b2 = beatapp.Beat(loglevel=logging.DEBUG)
         self.assertEqual(b2.loglevel, logging.DEBUG)
 
+    def test_colorize(self):
+        from celery import Celery
+        app = Celery(set_as_current=False)
+        app.log.setup = Mock()
+        b = beatapp.Beat(app=app, no_color=True)
+        b.setup_logging()
+        self.assertTrue(app.log.setup.called)
+        self.assertEqual(app.log.setup.call_args[1]['colorize'], False)
+
+
     def test_init_loader(self):
         b = beatapp.Beat()
         b.init_loader()

celery/tests/bin/test_worker.py

@@ -271,6 +271,7 @@ class test_Worker(WorkerAppCase):
         worker1 = self.Worker(loglevel=0xFFFF)
         self.assertEqual(worker1.loglevel, 0xFFFF)
 
+    @disable_stdouts
     def test_warns_if_running_as_privileged_user(self):
         app = current_app
         if app.IS_WINDOWS:
@@ -295,6 +296,31 @@ class test_Worker(WorkerAppCase):
         with self.assertRaises(AttributeError):
             sys.stdout.logger
 
+    @disable_stdouts
+    def test_on_start_custom_logging(self):
+        prev, self.app.log.redirect_stdouts = (
+            self.app.log.redirect_stdouts, Mock(),
+        )
+        try:
+            worker = self.Worker(redirect_stoutds=True)
+            worker._custom_logging = True
+            worker.on_start()
+            self.assertFalse(self.app.log.redirect_stdouts.called)
+        finally:
+            self.app.log.redirect_stdouts = prev
+
+    def test_setup_logging_no_color(self):
+        worker = self.Worker(redirect_stdouts=False, no_color=True)
+        prev, self.app.log.setup = self.app.log.setup, Mock()
+        worker.setup_logging()
+        self.assertFalse(self.app.log.setup.call_args[1]['colorize'])
+
+    @disable_stdouts
+    def test_startup_info_pool_is_str(self):
+        worker = self.Worker(redirect_stdouts=False)
+        worker.pool_cls = 'foo'
+        worker.startup_info()
+
     def test_redirect_stdouts_already_handled(self):
         logging_setup = [False]