Add tests for queue predeclaration.
import tempfile
+import mock
+from oslo_config import cfg
+
from st2common import service_setup
+from st2common.transport.bootstrap_utils import register_exchanges
+from st2common.transport.bootstrap_utils import QUEUES
from st2tests.base import CleanFilesTestCase
from st2tests import config
@@ -87,3 +92,21 @@ class ServiceSetupTestCase(CleanFilesTestCase):
register_signal_handlers=False,
register_internal_trigger_types=False,
run_migrations=False)
+
+ @mock.patch('kombu.Queue.declare')
+ def test_register_exchanges_predeclare_queues(self, mock_declare):
+ # Verify that queues are correctly pre-declared if the corresponding config option is set
+
+ # Pre-declaration is disabled
+ self.assertEqual(mock_declare.call_count, 0)
+ cfg.CONF.set_override(group='messaging', name='predeclare_queues', override=False)
+
+ register_exchanges()
+ self.assertEqual(mock_declare.call_count, 0)
+
+ # Pre-declaration is enabled
+ self.assertEqual(mock_declare.call_count, 0)
+ cfg.CONF.set_override(group='messaging', name='predeclare_queues', override=True)
+
+ register_exchanges()
+ self.assertEqual(mock_declare.call_count, len(QUEUES))
