Merge d0dad2b into 8c682b7

resultsdb/messaging.py

@@ -21,6 +21,7 @@
 import json
 
 import pkg_resources
+import stomp
 
 from resultsdb import db, app
 from resultsdb.models.results import Result, ResultData
@@ -176,9 +177,26 @@ def publish(self, message):
 
 class StompPlugin(MessagingPlugin):
     def __init__(self, **kwargs):
-        # Ensure that we can import this at startup time.
-        import stomp
-        self.stomp = stomp
+        conn_args = kwargs.get('connection', {})
+        if 'use_ssl' in conn_args:
+            use_ssl = conn_args['use_ssl']
+            del conn_args['use_ssl']
+        else:
+            use_ssl = False
+
+        ssl_args = {'for_hosts': conn_args.get('host_and_ports', [])}
+        for attr in ('key_file', 'cert_file', 'ca_certs'):
+            conn_attr = f'ssl_{attr}'
+            if conn_attr in conn_args:
+                ssl_args[attr] = conn_args[conn_attr]
+                del conn_args[conn_attr]
+
+        if 'ssl_version' in conn_args:
+            ssl_args['ssl_version'] = conn_args['ssl_version']
+            del conn_args['ssl_version']
+
+        kwargs['use_ssl'] = use_ssl
+        kwargs['ssl_args'] = ssl_args
 
         super(StompPlugin, self).__init__(**kwargs)
 
@@ -192,10 +210,11 @@ def publish(self, msg):
         msg = json.dumps(msg)
         kwargs = dict(body=msg, headers={}, destination=self.destination)
 
-        if self.stomp.__version__[0] < 4:
-            kwargs['message'] = kwargs.pop('body')  # On EL7, different sig.
+        conn = stomp.connect.StompConnection11(**self.connection)
+
+        if self.use_ssl:
+            conn.set_ssl(**self.ssl_args)
 
-        conn = self.stomp.Connection(**self.connection)
         conn.connect(wait=True)
         try:
             conn.send(**kwargs)

testing/test_general.py

@@ -1,4 +1,5 @@
 import datetime
+import ssl
 
 import resultsdb.controllers.api_v2 as apiv2
 import resultsdb.messaging as messaging
@@ -120,7 +121,7 @@ def test_load_plugin(self):
             plugin = messaging.load_messaging_plugin('fedmsg', {})
         except KeyError as err:
             if "not found" in str(err):
-                print ("""=============== HINT ===============
+                print("""=============== HINT ===============
 This exception can be caused by the fact, that you did not run
 `python setup.py develop` before executing the testsuite.
 
@@ -132,7 +133,47 @@ def test_load_plugin(self):
  - you might me missing the 'fedmsg' entrypoint in setup.py
  - there can be an error in the plugin loading code""")
             raise
-        assert isinstance(plugin, messaging.FedmsgPlugin), "check whether `fedmsg` entrypoint in setup.py points to resultsdb.messaging:FedmsgPlugin"
+        assert isinstance(plugin, messaging.FedmsgPlugin), (
+            "check whether `fedmsg` entrypoint in setup.py points to"
+            " resultsdb.messaging:FedmsgPlugin"
+        )
+
+    def test_load_stomp(self):
+        message_bus_kwargs = {
+            'destination': 'results.new',
+            'connection': {
+                'host_and_ports': [('localhost', 1234)],
+            },
+        }
+        plugin = messaging.load_messaging_plugin('stomp', message_bus_kwargs)
+        assert isinstance(plugin, messaging.StompPlugin)
+        assert plugin.destination == 'results.new'
+
+    def test_stomp_ssl(self):
+        message_bus_kwargs = {
+            'destination': 'results.new',
+            'connection': {
+                'host_and_ports': [('localhost', 1234)],
+
+                'use_ssl': True,
+                'ssl_version': ssl.PROTOCOL_TLSv1_2,
+                'ssl_key_file': '/etc/secret/umb-client.key',
+                'ssl_cert_file': '/etc/secret/umb-client.crt',
+                'ssl_ca_certs': '/etc/secret/ca.pem'
+            },
+        }
+        plugin = messaging.load_messaging_plugin('stomp', message_bus_kwargs)
+        assert plugin.connection == {
+            'host_and_ports': [('localhost', 1234)],
+        }
+        assert plugin.use_ssl is True
+        assert plugin.ssl_args == {
+            'for_hosts': [('localhost', 1234)],
+            'key_file': '/etc/secret/umb-client.key',
+            'cert_file': '/etc/secret/umb-client.crt',
+            'ca_certs': '/etc/secret/ca.pem',
+            'ssl_version': ssl.PROTOCOL_TLSv1_2,
+        }
 
 
 class TestGetResultsParseArgs():