Reproduce bug 1917645

The [oslo_messaging_notification]retry parameter is not applied during
connecting to the message bus. But the documentation implies it should[1][2].
The two possible drivers, rabbit and kafka, behaves differently.

1) The rabbit driver will retry the connection forever, blocking the caller
   process.

2) The kafka driver also ignores the retry configuration but the
   notifier call returns immediately even if the notification is not
   (cannot) be delivered.

This patch adds test cases to show the wrong behavior.

[1] https://docs.openstack.org/oslo.messaging/latest/configuration/opts.html#oslo_messaging_notifications.retry
[2] https://github.com/openstack/oslo.messaging/blob/feb72de7b81e3919dedc697f9fb5484a92f85ad8/oslo_messaging/notify/messaging.py#L31-L36

Related-Bug: #1917645

Change-Id: Id8557050157aecd3abd75c9114d3fcaecdfc5dc9
(cherry picked from commit 1db6de63a86812742cbc37a0f5fe1fd7a095dd7f)
(cherry picked from commit 7390034e479c044d9067d97cd801f9f58c813e41)

1 files changed, 68 insertions, 0 deletions

diff --git a/oslo_messaging/tests/notify/test_notifier.py b/oslo_messaging/tests/notify/test_notifier.py
index c36a432..d0a8eca 100644
--- a/oslo_messaging/tests/notify/test_notifier.py
+++ b/oslo_messaging/tests/notify/test_notifier.py
@@ -18,6 +18,7 @@ import sys
 import uuid
 
 import fixtures
+from kombu import connection
 from oslo_serialization import jsonutils
 from oslo_utils import strutils
 from oslo_utils import timeutils
@@ -228,6 +229,73 @@ class TestMessagingNotifier(test_utils.BaseTestCase):
 TestMessagingNotifier.generate_scenarios()
 
 
+class TestMessagingNotifierRetry(test_utils.BaseTestCase):
+
+    class TestingException(BaseException):
+        pass
+
+    def test_notifier_retry_connection_fails_rabbit(self):
+        """This test sets a small retry number for notification sending and
+        configures a non reachable message bus. The expectation that after the
+        configured number of retries the driver gives up the message sending.
+        """
+        self.config(
+            driver=["messagingv2"],
+            topics=["test-retry"],
+            retry=2,
+            group="oslo_messaging_notifications")
+        transport = oslo_messaging.get_notification_transport(
+            self.conf, url='rabbit://')
+        notifier = oslo_messaging.Notifier(transport)
+
+        orig_establish_connection = connection.Connection._establish_connection
+        calls = []
+
+        def wrapped_establish_connection(*args, **kwargs):
+            if len(calls) > 2:
+                raise self.TestingException(
+                    "Connection should only be retried twice due to "
+                    "configuration")
+            else:
+                calls.append((args, kwargs))
+                orig_establish_connection(*args, **kwargs)
+
+        with mock.patch(
+            'kombu.connection.Connection._establish_connection',
+            new=wrapped_establish_connection
+        ):
+            # FIXME(gibi) This is bug 1917645 as the driver does not stop
+            # retrying the connection after two retries only our test fixture
+            # stops the retry by raising TestingException
+            self.assertRaises(
+                self.TestingException,
+                notifier.info, {}, "test", {})
+
+    def test_notifier_retry_connection_fails_kafka(self):
+        """This test sets a small retry number for notification sending and
+        configures a non reachable message bus. The expectation that after the
+        configured number of retries the driver gives up the message sending.
+        """
+
+        self.config(
+            driver=["messagingv2"],
+            topics=["test-retry"],
+            retry=2,
+            group='oslo_messaging_notifications')
+
+        transport = oslo_messaging.get_notification_transport(
+            self.conf, url='kafka://')
+
+        notifier = oslo_messaging.Notifier(transport)
+
+        # Kafka's message producer interface is async, and there is no way
+        # from the oslo interface to force sending a pending message. So this
+        # call simply returns without i) failing to deliver the message to
+        # the non existent kafka bus ii) retrying the message delivery twice
+        # as the configuration requested it.
+        notifier.info({}, "test", {})
+
+
 class TestSerializer(test_utils.BaseTestCase):
 
     def setUp(self):