Merge lp:~adeuring/lazr.jobrunner/bug1015667-2 into lp:lazr.jobrunner

[Abel Deuring]

This branch makes the lazr.jobrunner script usable with the Celery
configuration used by Launchpad.

The problem: LP's Celery configuration sets CELERY_CREATE_MISSING_QUEUES
to False to avoid the creation of arbitrary queues. With this setting,
an attempt to create an instance of app.amqp.Router in the function
drain_queues() fails with the error "queue not found in CELERY_QUEUES".

This error can be avoid by calling app.amqp.Router(create_missing=True).

Attempts to write a test for this change revealed a test isolation problem
in the existing tests of clear_queues(): The est modules simply imported
the function clear_queues() from the module bin.clear_queues, and
clear_queues() does a "late import" of lazr.jobrunner.celerytask, which
in turn imports indirectly a Celery config module.

This meant that the Celery configuration of the first test was used
in all subsequent tests.

I changed the tests so that the tests now invoke the real script in a
subprocess.

This is done via the context manager running(), which tried to call
proc.terminate(). This failed with an exception for the clear-queues
script because the script is already terminated when proc.terminate()
is called. The call of proc.terminate() is now optional in running().

Finally, I refactored to Celery test configurations a bit: We should
always parameters like BROKER_VHOST or CELERY_RESULT_BACKEND, but
overall quite simple the configuration "config1" needs an environment
variable FILE_JOB_DIR, which is unnecessary for the tests of
clear-queues, so I added a quite basic configuration "simple_config",
which is imported by config1 and by the second new configuration
config_do_not_create_missing_queues.

test: make check

no lint

[lp:~adeuring/lazr.jobrunner/bug1015667-2]

44. By Abel Deuring on 2012-07-09

use Popen() directly in runClearQueues()

[Aaron Bentley (abentley)]

review: Approve

=== modified file 'src/lazr/jobrunner/celerytask.py'
--- src/lazr/jobrunner/celerytask.py	2012-07-03 14:09:46 +0000
+++ src/lazr/jobrunner/celerytask.py	2012-07-09 15:57:22 +0000
@@ -95,7 +95,7 @@
     if callbacks is None:
         callbacks = [lambda x, y: None]
     bindings = []
-    router = app.amqp.Router()
+    router = app.amqp.Router(create_missing=True)
     for queue_name in queue_names:
         destination = router.expand_destination(queue_name)
         exchange = Exchange(destination['exchange'])
@@ -111,8 +111,9 @@
             # This is basically copied from kombu.Queue.declare().
             # We can't use this method directly because queue_declare()
             # must be called with passive=True for result queues.
-            # Otherwise, attempts to connect to the queue fail with
-            # AMQPChannelException: (406, u"PRECONDITION_FAILED...", ...)
+            # Otherwise, attempts to connect to the queue fails with
+            # celery.exceptions.QueueNotFound: "Queue ... is not defined
+            # in CELERY_QUEUES".
             for queue in consumer.queues:
                 if queue.exchange:
                     queue.exchange.declare()

=== modified file 'src/lazr/jobrunner/tests/config1.py'
--- src/lazr/jobrunner/tests/config1.py	2012-03-21 20:38:50 +0000
+++ src/lazr/jobrunner/tests/config1.py	2012-07-09 15:57:22 +0000
@@ -1,7 +1,4 @@
-BROKER_VHOST = "/"
-CELERY_RESULT_BACKEND = "amqp"
-CELERY_IMPORTS = ("lazr.jobrunner.tests.test_celerytask", )
-CELERYD_CONCURRENCY = 1
+from simple_config import *
 import os
 import oops
 CELERY_ANNOTATIONS = {

=== added file 'src/lazr/jobrunner/tests/config_do_not_create_missing_queues.py'
--- src/lazr/jobrunner/tests/config_do_not_create_missing_queues.py	1970-01-01 00:00:00 +0000
+++ src/lazr/jobrunner/tests/config_do_not_create_missing_queues.py	2012-07-09 15:57:22 +0000
@@ -0,0 +1,2 @@
+from simple_config import *
+CELERY_CREATE_MISSING_QUEUES = False

=== added file 'src/lazr/jobrunner/tests/simple_config.py'
--- src/lazr/jobrunner/tests/simple_config.py	1970-01-01 00:00:00 +0000
+++ src/lazr/jobrunner/tests/simple_config.py	2012-07-09 15:57:22 +0000
@@ -0,0 +1,4 @@
+BROKER_VHOST = "/"
+CELERY_RESULT_BACKEND = "amqp"
+CELERY_IMPORTS = ("lazr.jobrunner.tests.test_celerytask", )
+CELERYD_CONCURRENCY = 1

=== modified file 'src/lazr/jobrunner/tests/test_celerytask.py'
--- src/lazr/jobrunner/tests/test_celerytask.py	2012-07-04 16:34:25 +0000
+++ src/lazr/jobrunner/tests/test_celerytask.py	2012-07-09 15:57:22 +0000
@@ -18,7 +18,6 @@
 
 
 import contextlib
-from cStringIO import StringIO
 import errno
 import json
 import os
@@ -29,7 +28,6 @@
     )
 import shutil
 import subprocess
-import sys
 import tempfile
 from time import sleep
 from unittest import TestCase
@@ -39,7 +37,6 @@
 
 from celery.exceptions import SoftTimeLimitExceeded
 
-from lazr.jobrunner.bin.clear_queues import clear_queues
 from lazr.jobrunner.celerytask import (
     drain_queues,
     list_queued,
@@ -481,30 +478,28 @@
         self.assertEqual([], list_queued(RunFileJob.app, [self.queue]))
 
 
-class TestInspectQueues(TestCase):
+class TestClearQueues(TestCase):
     """Tests for the script inspect-queues."""
 
     def queueName(self, task_id):
         return task_id.replace('-', '')
 
-    def runInspectQueues(self, celery_config, task_ids):
+    def runClearQueues(self, celery_config, task_ids):
         """Invoke clear_queues() and catch the data written to stdout
         and stderr.
         """
+        # Simply calling the function clear_queues() from bin.clear_queues()
+        # leads to a one-time import of the celery config module; the
+        # config setting from the test that runs first would override
+        # any different configuration setting in a later test.
+        # Running the script in a subprocess avoids this problem.
         queues = [self.queueName(task_id) for task_id in task_ids]
-        real_stdout = sys.stdout
-        real_stderr = sys.stderr
-        try:
-            sys.stdout = StringIO()
-            sys.stderr = StringIO()
-            args = ['program', '-c', celery_config] + queues
-            clear_queues(args)
-            fake_stdout = sys.stdout.getvalue()
-            fake_stderr = sys.stderr.getvalue()
-        finally:
-            sys.stdout = real_stdout
-            sys.stderr = real_stderr
-        return fake_stdout, fake_stderr
+        args = ['bin/clear-queues', '-c', celery_config] + queues
+        proc = subprocess.Popen(
+            args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+            cwd=get_root())
+        stdout, stderr = proc.communicate()
+        return stdout, stderr
 
     def invokeJob(self, celery_config, task, delay=1, job_args={}):
         """Run the given task.
@@ -537,15 +532,19 @@
         but the result is not consumed, the related message can be
         retrieved with clear_queues().
         """
-        celery_config = 'lazr.jobrunner.tests.config1'
-        task_id = self.invokeJob(celery_config, RunFileJob)
-        stdout, stderr = self.runInspectQueues(celery_config, [task_id])
+        celery_config_jobrunner = 'lazr.jobrunner.tests.config1'
+        task_id = self.invokeJob(celery_config_jobrunner, RunFileJob)
+        # The script clear_queues does not have to use the same
+        # Celery configuration as the job runner: clear_config just
+        # needs to know how to connect to AMQP server.
+        clear_queue_config = 'lazr.jobrunner.tests.simple_config'
+        stdout, stderr = self.runClearQueues(clear_queue_config, [task_id])
         self.assertEqual(self.successMessage(task_id), stdout)
         self.assertEqual('', stderr)
 
         # Reading a queue is destructive. An attempt to read again from
         # a queue results in an error.
-        stdout, stderr = self.runInspectQueues(celery_config, [task_id])
+        stdout, stderr = self.runClearQueues(clear_queue_config, [task_id])
         self.assertEqual('', stdout)
         self.assertEqual(self.noQueueMessage(task_id), stderr)
 
@@ -553,11 +552,12 @@
         """More than one queue can be inspected in one call of
         clear_queue().
         """
-        celery_config = 'lazr.jobrunner.tests.config1'
-        task_id_1 = self.invokeJob(celery_config, RunFileJob)
-        task_id_2 = self.invokeJob(celery_config, RunFileJob)
-        stdout, stderr = self.runInspectQueues(
-            celery_config, [task_id_1, task_id_2])
+        celery_config_jobrunner = 'lazr.jobrunner.tests.config1'
+        task_id_1 = self.invokeJob(celery_config_jobrunner, RunFileJob)
+        task_id_2 = self.invokeJob(celery_config_jobrunner, RunFileJob)
+        clear_queue_config = 'lazr.jobrunner.tests.simple_config'
+        stdout, stderr = self.runClearQueues(
+            clear_queue_config, [task_id_1, task_id_2])
         expected_stdout = (
             self.successMessage(task_id_1) + self.successMessage(task_id_2))
         self.assertEqual(expected_stdout, stdout)
@@ -567,8 +567,28 @@
         """A Celery task which was started so that no result is returned
         does not write to a task queue.
         """
-        celery_config = 'lazr.jobrunner.tests.config1'
-        task_id = self.invokeJob(celery_config, RunFileJobNoResult)
-        stdout, stderr = self.runInspectQueues(celery_config, [task_id])
+        celery_config_jobrunner = 'lazr.jobrunner.tests.config1'
+        task_id = self.invokeJob(celery_config_jobrunner, RunFileJobNoResult)
+        clear_queue_config = 'lazr.jobrunner.tests.simple_config'
+        stdout, stderr = self.runClearQueues(clear_queue_config, [task_id])
+        self.assertEqual('', stdout)
+        self.assertEqual(self.noQueueMessage(task_id), stderr)
+
+    def test_clear_queues__config_create_missing_queues_false(self):
+        """If CELERY_CREATE_MISSING_QUEUES is False, drain_queues()
+        must override this setting by creating the router instance
+        with the parameter create_missing=True. Otherwise,
+        app.amqp.Router() will fail with
+        'celery.exceptions.QueueNotFound: "Queue ... is not defined in
+        CELERY_QUEUES"'. Note that this does not mean that a non-existent
+        queue is actually created, see
+        assertEqual(self.noQueueMessage(task_id), stderr) below...
+        """
+        celery_config = (
+            'lazr.jobrunner.tests.config_do_not_create_missing_queues')
+        # A test isolation problem: Specifying a configuration module does
+        # not have the desired effect
+        task_id = 'this-queue-does-not-exist'
+        stdout, stderr = self.runClearQueues(celery_config, [task_id])
         self.assertEqual('', stdout)
         self.assertEqual(self.noQueueMessage(task_id), stderr)