Merge lp:~thomir-deactivatedaccount/core-image-publisher/trunk-add-queue-listener into lp:core-image-publisher

[Thomi Richards]

Add code to read from amqp.

Add code to connect to, and read from amqp. A few changes from last sprint, in order to support testability.

[Celso Providelo (cprov)]

Thomi,

I suppose we can use the on_message callback pattern for consumers, It kind of adds some additional indirection levels but makes it testable.

Let's got with it till the completion of the publisher, then we can evaluate if it's good to be propagated to the other services.

review: Approve

=== modified file 'core_image_publisher/__init__.py'
--- core_image_publisher/__init__.py	2015-03-25 02:46:02 +0000
+++ core_image_publisher/__init__.py	2015-03-25 21:37:19 +0000
@@ -1,4 +1,19 @@
-
+# core-image-publisher
+# Copyright (C) 2015 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
 
 import select
 

=== added file 'core_image_publisher/constants.py'
--- core_image_publisher/constants.py	1970-01-01 00:00:00 +0000
+++ core_image_publisher/constants.py	2015-03-25 21:37:19 +0000
@@ -0,0 +1,20 @@
+# core-image-publisher
+# Copyright (C) 2015 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+"""Constants for this service."""
+
+API_VERSION="v1"

=== added file 'core_image_publisher/queue.py'
--- core_image_publisher/queue.py	1970-01-01 00:00:00 +0000
+++ core_image_publisher/queue.py	2015-03-25 21:37:19 +0000
@@ -0,0 +1,78 @@
+# core-image-publisher
+# Copyright (C) 2015 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import logging
+
+import kombu
+from kombu.mixins import ConsumerMixin
+
+from core_image_publisher import constants
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    'CoreImagePublisherQueueMonitor'
+]
+
+
+class CoreImagePublisherQueueMonitor(ConsumerMixin):
+
+    """A class that watches for incoming messages from a queue, and calls
+    a callback to process them.
+
+    By using an external callback we can separate the stateful logic of
+    connecting to amqp from our business logic of dealing with the messages.
+
+    """
+
+    def __init__(self, connection, on_message_cb):
+        """Construct a new CoreImagePublisherQueueMonitor.
+
+        :param connection: A kombu.Connection instance.
+        :param on_message_db: A callable that will be called for every new
+            messages from the rabbitMQ queue(s). It must accept a single
+            parameter: the message object. The callback MUST either acknowledge
+            or reject or requeue the messsage, and MUST NOT raise any unhandled
+            exceptions.
+
+        """
+        self.connection = connection
+        self.on_message_cb = on_message_cb
+
+    def get_consumers(self, Consumer, channel):
+        """Return consumers instances for all configured queues."""
+        queues = [
+            kombu.Queue('core.image.{}'.format(constants.API_VERSION))
+        ]
+        return [Consumer(queues=queues, callbacks=[self.process])]
+
+    def process(self, body, message):
+        """Process incomming test request.
+
+        Run requested tests and posts results to the 'adt_results' queue
+        for later checking.
+        """
+        logger.info('Got: {}'.format(body), extra=body)
+        try:
+            self.on_message_cb(message)
+        except Exception as e:
+            logger.error(
+                "Caught unhandled exception while processing message: %s",
+                e,
+                extra=message.payload
+            )
+            message.requeue()

=== added directory 'core_image_publisher/tests'
=== added file 'core_image_publisher/tests/__init__.py'
--- core_image_publisher/tests/__init__.py	1970-01-01 00:00:00 +0000
+++ core_image_publisher/tests/__init__.py	2015-03-25 21:37:19 +0000
@@ -0,0 +1,16 @@
+# core-image-publisher
+# Copyright (C) 2015 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#

=== added file 'core_image_publisher/tests/test_queue_integration.py'
--- core_image_publisher/tests/test_queue_integration.py	1970-01-01 00:00:00 +0000
+++ core_image_publisher/tests/test_queue_integration.py	2015-03-25 21:37:19 +0000
@@ -0,0 +1,76 @@
+# core-image-publisher
+# Copyright (C) 2015 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+"""Integration tests for the kombu queue code."""
+
+import kombu
+import kombu.simple
+from testtools import TestCase
+
+
+from core_image_publisher.queue import CoreImagePublisherQueueMonitor
+
+
+class KombuQueueIntegrationTests(TestCase):
+
+    def test_can_read_message(self):
+        conn = kombu.Connection('memory:///')
+        queue_message(conn, {'test': 'value'})
+
+        consumer = LoggingConsumer()
+        q = CoreImagePublisherQueueMonitor(conn, consumer)
+        # pump the queue to get the enqueued message:
+        next(q.consume(limit=1, timeout=5.0))
+
+        self.assertEqual(consumer.got_messages, [dict(test='value')])
+
+    def test_uncaught_exceptions_cause_message_requeue(self):
+        conn = kombu.Connection('memory:///')
+        queue_message(conn, {'test': 'value'})
+
+        def consumer_with_bug(message):
+            raise RuntimeError("I am a bug, all up in ur bizniz!")
+
+        logging_consumer = LoggingConsumer()
+
+        queue_bugged = CoreImagePublisherQueueMonitor(conn, consumer_with_bug)
+        queue_ok = CoreImagePublisherQueueMonitor(conn, logging_consumer)
+
+        # pump the queue to get the enqueued message:
+        next(queue_bugged.consume(limit=1, timeout=5.0))
+        next(queue_ok.consume(limit=1, timeout=5.0))
+
+        self.assertEqual(logging_consumer.got_messages, [dict(test='value')])
+
+
+
+class LoggingConsumer(object):
+
+    """A consumer callback object that acks and logs all received payloads."""
+
+    def __init__(self):
+        self.got_messages = []
+
+    def __call__(self, message):
+        self.got_messages.append(message.payload)
+        message.ack()
+
+
+def queue_message(conn, message):
+    q = kombu.simple.SimpleQueue(conn, 'core.image.v1')
+    q.put(message)
+

=== modified file 'requirements.txt'
--- requirements.txt	2015-03-25 02:46:02 +0000
+++ requirements.txt	2015-03-25 21:37:19 +0000
@@ -0,0 +1,1 @@
+kombu==3.0.24
\ No newline at end of file

=== modified file 'setup.py'
--- setup.py	2015-03-25 04:19:49 +0000
+++ setup.py	2015-03-25 21:37:19 +0000
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-# core-image-watcher 
+# core-image-publisher
 # Copyright (C) 2015 Canonical
 #
 # This program is free software: you can redistribute it and/or modify
@@ -36,5 +36,6 @@
     url='https://launchpad.net/core-image-publisher',
     license='GPLv3',
     packages=find_packages(),
-    scripts=['core-image-publisher.py']
+    scripts=['core-image-publisher.py'],
+    test_suite='core_image_publisher.tests'
 )

=== modified file 'test_requirements.txt'
--- test_requirements.txt	2015-03-25 02:46:02 +0000
+++ test_requirements.txt	2015-03-25 21:37:19 +0000
@@ -0,0 +1,1 @@
+testtools==1.7.1
\ No newline at end of file