100% Coverage for celery.bin.celeryd, .celerybeat, .base

celery/bin/base.py

@@ -80,6 +80,7 @@ class Command(object):
             self.app = self.get_cls_by_name(app)
         else:
             self.app = self.get_app(loader=loader)
+        print("SETUP")
         if self.enable_config_from_cmdline:
             argv = self.process_cmdline_config(argv)
         return argv

celery/loaders/base.py

@@ -83,8 +83,8 @@ class BaseLoader(object):
 
             ## find key/value
             # ns.key=value|ns_key=value (case insensitive)
-            key, value = arg.replace('.', '_').split('=', 1)
-            key = key.upper()
+            key, value = arg.split('=', 1)
+            key = key.upper().replace(".", "_")
 
             ## find namespace.
             # .key=value|_key=value expands to default namespace.

celery/tests/test_bin/test_base.py

@@ -1,12 +1,20 @@
-from celery.tests.utils import unittest
+import os
 
 from celery.bin.base import Command
 
+from celery.tests.utils import unittest
+
 
 class Object(object):
     pass
 
 
+class MyApp(object):
+    pass
+
+APP = MyApp()  # <-- Used by test_with_custom_app
+
+
 class MockCommand(Command):
 
     def parse_options(self, prog_name, arguments):
@@ -39,3 +47,32 @@ class test_Command(unittest.TestCase):
         self.assertTupleEqual(args2, (10, 20, 30))
         self.assertDictContainsSubset({"foo": "bar", "prog_name": "foo"},
                                       kwargs2)
+
+    def test_with_custom_config_module(self):
+        prev = os.environ.pop("CELERY_CONFIG_MODULE", None)
+        try:
+            cmd = MockCommand()
+            cmd.setup_app_from_commandline(["--config=foo.bar.baz"])
+            self.assertEqual(os.environ.get("CELERY_CONFIG_MODULE"),
+                             "foo.bar.baz")
+        finally:
+            if prev:
+                os.environ["CELERY_CONFIG_MODULE"] = prev
+
+    def test_with_custom_app(self):
+        cmd = MockCommand()
+        app = ".".join([__name__, "APP"])
+        cmd.setup_app_from_commandline(["--app=%s" % (app, ),
+                                        "--loglevel=INFO"])
+        self.assertIs(cmd.app, APP)
+
+    def test_with_cmdline_config(self):
+        cmd = MockCommand()
+        cmd.enable_config_from_cmdline = True
+        cmd.namespace = "celeryd"
+        rest = cmd.setup_app_from_commandline(argv=[
+            "--loglevel=INFO", "--", "broker.host=broker.example.com",
+            ".prefetch_multiplier=100"])
+        self.assertEqual(cmd.app.conf.BROKER_HOST, "broker.example.com")
+        self.assertEqual(cmd.app.conf.CELERYD_PREFETCH_MULTIPLIER, 100)
+        self.assertListEqual(rest, ["--loglevel=INFO"])

celery/tests/test_bin/test_celerybeat.py

@@ -155,10 +155,28 @@ class test_Beat(unittest.TestCase):
             platforms.create_pidlock = prev
 
 
+class MockDaemonContext(object):
+    opened = False
+    closed = False
+
+    def open(self):
+        self.__class__.opened = True
+
+    def close(self):
+        self.__class__.closed = True
+
+
+def create_daemon_context(*args, **kwargs):
+    context = MockDaemonContext()
+    return context, context.close
+
+
 class test_div(unittest.TestCase):
 
     def setUp(self):
         self.prev, beatapp.Beat = beatapp.Beat, MockBeat
+        self.ctx, celerybeat_bin.create_daemon_context = \
+                celerybeat_bin.create_daemon_context, create_daemon_context
 
     def tearDown(self):
         beatapp.Beat = self.prev
@@ -171,6 +189,13 @@ class test_div(unittest.TestCase):
         finally:
             MockBeat.running = False
 
+    def test_detach(self):
+        cmd = celerybeat_bin.BeatCommand()
+        cmd.app = app_or_default()
+        cmd.run(detach=True)
+        self.assertTrue(MockDaemonContext.opened)
+        self.assertTrue(MockDaemonContext.closed)
+
     def test_parse_options(self):
         cmd = celerybeat_bin.BeatCommand()
         cmd.app = app_or_default()

celery/tests/test_bin/test_celeryd.py

@@ -4,12 +4,15 @@ import sys
 
 from multiprocessing import get_logger, current_process
 
+from kombu.tests.utils import redirect_stdouts
+
 from celery import Celery
 from celery import platforms
 from celery import signals
 from celery.app import app_or_default
 from celery.apps import worker as cd
-from celery.bin.celeryd import WorkerCommand, main as celeryd_main
+from celery.bin.celeryd import WorkerCommand, windows_main, \
+                               main as celeryd_main
 from celery.exceptions import ImproperlyConfigured
 from celery.utils import patch
 from celery.utils.functional import wraps
@@ -312,6 +315,12 @@ class test_Worker(unittest.TestCase):
 
 class test_funs(unittest.TestCase):
 
+    @redirect_stdouts
+    def test_windows_main(self, stdout, stderr):
+        windows_main()
+        self.assertIn("celeryd command does not work on Windows",
+                      stderr.getvalue())
+
     @disable_stdouts
     def test_set_process_status(self):
         worker = Worker(hostname="xyzza")