Port to fedora-messaging

Fedora infrastructure is going to decommission the fedmsg-gateway
endpoint we've been consuming, which requires us to migrate off that
stack entirely.

requirements.txt

@@ -3,6 +3,7 @@ jira
 requests
 requests_kerberos
 fedmsg
+fedora-messaging
 PyGithub
 pypandoc_binary
 urllib3

sync2jira/compat.py

@@ -4,10 +4,8 @@ def extract_message_body(msg):
     Handle both fedmsg and fedora-messaging style message bodies.
     """
 
-    body = msg.get("msg", {})
-    body = body.get("body", body)
+    body = msg.get("body", {})
     if body:
         return body
-    raise KeyError(
-        f"Unrecognized message format with keys {msg.keys()}. Expected either 'msg' or 'body'"
-    )
+    else:
+        return msg

sync2jira/main.py

@@ -17,11 +17,7 @@
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110.15.0 USA
 #
 # Authors:  Ralph Bean <[email protected]>
-"""Sync GitHub issues to a jira instance, via fedmsg.
-
-Run with systemd, please.
-"""
-from copy import deepcopy
+"""Sync GitHub issues to a jira instance, via fedora-messaging."""
 
 # Build-In Modules
 import logging
@@ -31,8 +27,8 @@
 import warnings
 
 # 3rd Party Modules
-import fedmsg
 import fedmsg.config
+import fedora_messaging.api
 import jinja2
 
 # Local Modules
@@ -146,6 +142,22 @@ def load_config(loader=fedmsg.config.load_config):
     return config
 
 
+def callback(msg):
+    config = load_config()
+    topic = msg.topic
+    idx = msg.id
+    suffix = ".".join(topic.split(".")[3:])
+    log.debug("Encountered %r %r %r", suffix, topic, idx)
+
+    if suffix not in issue_handlers and suffix not in pr_handlers:
+        return
+
+    log.debug("Handling %r %r %r", suffix, topic, idx)
+
+    body = c.extract_message_body(msg.body)
+    handle_msg(body, suffix, config)
+
+
 def listen(config):
     """
     Listens to activity on upstream repos on GitHub
@@ -159,19 +171,33 @@ def listen(config):
         log.info("`listen` is disabled.  Exiting.")
         return
 
-    log.info("Waiting for a relevant fedmsg message to arrive...")
-    for _, _, topic, msg in fedmsg.tail_messages(**config):
-        idx = msg["msg_id"]
-        suffix = ".".join(topic.split(".")[3:])
-        log.debug("Encountered %r %r %r", suffix, topic, idx)
-
-        if suffix not in issue_handlers and suffix not in pr_handlers:
-            continue
-
-        log.debug("Handling %r %r %r", suffix, topic, idx)
+    # Next, we need a queue to consume messages from. We can define
+    # the queue and binding configurations in these dictionaries:
+    queue = os.getenv("FEDORA_MESSAGING_QUEUE", "8b16c196-7ee3-4e33-92b9-e69d80fce333")
+    queues = {
+        queue: {
+            "durable": True,  # Persist the queue on broker restart
+            "auto_delete": False,  # Delete the queue when the client terminates
+            "exclusive": True,  # Disallow multiple simultaneous consumers
+            "arguments": {},
+        },
+    }
+    bindings = {
+        "exchange": "amq.topic",  # The AMQP exchange to bind our queue to
+        "queue": queue,
+        "routing_keys": [  # The topics that should be delivered to the queue
+            # New style
+            "org.fedoraproject.prod.github.issues",
+            "org.fedoraproject.prod.github.issue_comment",
+            "org.fedoraproject.prod.github.pull_request",
+            # Old style
+            "org.fedoraproject.prod.github.issue.#",
+            "org.fedoraproject.prod.github.pull_request.#",
+        ],
+    }
 
-        body = c.extract_message_body(msg)
-        handle_msg(body, suffix, config)
+    log.info("Waiting for a relevant fedmsg message to arrive...")
+    fedora_messaging.api.consume(callback, bindings=bindings, queues=queues)
 
 
 def initialize_issues(config, testing=False, repo_name=None):

tests/test_main.py

@@ -7,6 +7,13 @@
 PATH = "sync2jira.main."
 
 
+class MockMessage(object):
+    def __init__(self, msg_id, body, topic):
+        self.id = msg_id
+        self.body = body
+        self.topic = topic
+
+
 class TestMain(unittest.TestCase):
     """
     This class tests the main.py file under sync2jira
@@ -31,14 +38,16 @@ def setUp(self):
 
         # Mock Fedmsg Message
         self.mock_message_body = {"issue": "mock_issue"}
-        self.old_style_mock_message = {
-            "msg_id": "mock_id",
-            "msg": self.mock_message_body,
-        }
-        self.new_style_mock_message = {
-            "msg_id": "mock_id",
-            "msg": {"body": self.mock_message_body},
-        }
+        self.old_style_mock_message = MockMessage(
+            msg_id="mock_id",
+            body=self.mock_message_body,
+            topic=None,
+        )
+        self.new_style_mock_message = MockMessage(
+            msg_id="mock_id",
+            body={"body": self.mock_message_body},
+            topic=None,
+        )
 
     def _check_for_exception(self, loader, target, exc=ValueError):
         try:
@@ -283,67 +292,57 @@ def test_initialize_github_error(
         mock_report_failure.assert_called_with(self.mock_config)
 
     @mock.patch(PATH + "handle_msg")
-    @mock.patch(PATH + "fedmsg")
-    def test_listen_no_handlers(self, mock_fedmsg, mock_handle_msg):
+    @mock.patch(PATH + "load_config")
+    def test_listen_no_handlers(self, mock_load_config, mock_handle_msg):
         """
         Test 'listen' function where suffix is not in handlers
         """
         # Set up return values
-        mock_fedmsg.tail_messages.return_value = [
-            ("dummy", "dummy", "mock_topic", self.old_style_mock_message)
-        ]
+        mock_load_config.return_value = self.mock_config
 
         # Call the function
-        m.listen(self.mock_config)
+        self.old_style_mock_message.topic = "mock_topic"
+        m.callback(self.old_style_mock_message)
 
         # Assert everything was called correctly
         mock_handle_msg.assert_not_called()
 
     @mock.patch(PATH + "handle_msg")
     @mock.patch(PATH + "issue_handlers")
-    @mock.patch(PATH + "fedmsg")
-    def test_listen_no_issue(self, mock_fedmsg, mock_handlers_issue, mock_handle_msg):
+    @mock.patch(PATH + "load_config")
+    def test_listen_no_issue(
+        self, mock_load_config, mock_handlers_issue, mock_handle_msg
+    ):
         """
         Test 'listen' function where the handler returns none
         """
         # Set up return values
         mock_handlers_issue["github.issue.comment"].return_value = None
-        mock_fedmsg.tail_messages.return_value = [
-            ("dummy", "dummy", "d.d.d.github.issue.drop", self.old_style_mock_message)
-        ]
+        mock_load_config.return_value = self.mock_config
 
         # Call the function
-        m.listen(self.mock_config)
+        self.old_style_mock_message.topic = "d.d.d.github.issue.drop"
+        m.callback(self.old_style_mock_message)
 
         # Assert everything was called correctly
         mock_handle_msg.assert_not_called()
 
     @mock.patch(PATH + "handle_msg")
     @mock.patch(PATH + "issue_handlers")
-    @mock.patch(PATH + "fedmsg")
-    def test_listen(self, mock_fedmsg, mock_handlers_issue, mock_handle_msg):
+    @mock.patch(PATH + "load_config")
+    def test_listen(self, mock_load_config, mock_handlers_issue, mock_handle_msg):
         """
         Test 'listen' function where everything goes smoothly
         """
         # Set up return values
         mock_handlers_issue["github.issue.comment"].return_value = "dummy_issue"
-        mock_fedmsg.tail_messages.return_value = [
-            (
-                "dummy",
-                "dummy",
-                "d.d.d.github.issue.comment",
-                self.old_style_mock_message,
-            ),
-            (
-                "dummy",
-                "dummy",
-                "d.d.d.github.issue.comment",
-                self.new_style_mock_message,
-            ),
-        ]
+        mock_load_config.return_value = self.mock_config
 
         # Call the function
-        m.listen(self.mock_config)
+        self.old_style_mock_message.topic = "d.d.d.github.issue.comment"
+        self.new_style_mock_message.topic = "d.d.d.github.issue.comment"
+        m.callback(self.old_style_mock_message)
+        m.callback(self.new_style_mock_message)
 
         # Assert everything was called correctly
         # It should be called twice, once for the old style message and once for the new.