Add specific handling for RMQ checks without a configured vhost (#119)

Add test coverage for `check_rmq_is_available`
Update logging

neon_mq_connector/connector.py

@@ -576,6 +576,7 @@ def run_consumers(self, names: Optional[tuple] = None, daemon=True):
                 self.consumers[name].daemon = daemon
                 self.consumers[name].start()
                 self.consumer_properties[name]['started'] = True
+        LOG.debug(f"Started consumers for {self.service_name}")
 
     def stop_consumers(self, names: Optional[tuple] = None):
         """
@@ -596,6 +597,7 @@ def stop_consumers(self, names: Optional[tuple] = None):
                     self.consumer_properties[name]['started'] = False
             except Exception as e:
                 raise ChildProcessError(e)
+        LOG.debug(f"Stopped consumers for {self.service_name}")
 
     @retry(callback_on_exceeded='stop_sync_thread', use_self=True,
            num_retries=__run_retries__)
@@ -705,6 +707,7 @@ def stop(self):
         self.stop_sync_thread()
         self.stop_observer_thread()
         self._started = False
+        LOG.info(f"Stopped Connector {self.service_name}")
 
     def pre_run(self, **kwargs):
         """Additional logic invoked before method run()"""

neon_mq_connector/utils/connection_utils.py

@@ -27,14 +27,13 @@
 # SOFTWARE,  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 import logging
 import time
-from asyncio import IncompleteReadError
+
 from threading import Event
 from typing import Union, Callable, Optional
 from ovos_utils.log import LOG
 from pika.adapters.blocking_connection import BlockingConnection
-from pika.adapters.utils.connection_workflow import AMQPConnectionWorkflowFailed, AMQPConnectorException
 from pika.connection import ConnectionParameters
-from pika.exceptions import AMQPConnectionError, IncompatibleProtocolError
+from pika.exceptions import IncompatibleProtocolError, ProbableAccessDeniedError
 
 from neon_mq_connector.utils.network_utils import check_port_is_open
 
@@ -190,11 +189,15 @@ def check_rmq_is_available(
         connection = BlockingConnection(connection_params)
         connection.close()
         success = True
-    except AMQPConnectionError as e:
-        if isinstance(e, IncompatibleProtocolError):
-            LOG.debug(f"RMQ is likely still starting up (e={e})")
+    except IncompatibleProtocolError as e:
+        LOG.debug(f"RMQ is likely still starting up (e={e})")
+    except ProbableAccessDeniedError as e:
+        if connection_params.virtual_host == '/':
+            LOG.warning(f"Access was denied to default vhost='/'. Assuming RMQ "
+                        f"broker is online.")
+            success = True
         else:
             raise e
     finally:
         pika_log.setLevel(pika_level)
-        return success
+    return success

tests/test_utils.py

@@ -35,7 +35,9 @@
 import pytest
 import pika
 
-from threading import Thread, Event
+from threading import Thread
+
+from pika.exceptions import ProbableAuthenticationError
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
@@ -208,12 +210,23 @@ def test_connector_shutdown(self):
         self.assertTrue(connector.connection.is_closed)
 
 
+@pytest.mark.usefixtures("rmq_instance")
 class TestMQConnectionUtils(unittest.TestCase):
+    test_conf = None
     counter = 0
 
     def setUp(self) -> None:
         self.counter = 0
 
+        if self.test_conf is None:
+            self.test_conf = {
+                "server": "localhost",
+                "port": self.rmq_instance.port,
+                "users": {"mq_handler": {"user": "test_user",
+                                         "password": "test_password"}}}
+            import neon_mq_connector.utils.client_utils
+            neon_mq_connector.utils.client_utils._default_mq_config = self.test_conf
+
     def repeating_method(self):
         """Simple method incrementing counter by one"""
         self.counter += 1
@@ -260,7 +273,43 @@ def test_wait_for_mq_startup(self):
 
     def test_check_rmq_is_available(self):
         from neon_mq_connector.utils.connection_utils import check_rmq_is_available
-        # TODO
+        from pika.exceptions import ProbableAccessDeniedError
+        from pika.credentials import PlainCredentials
+        from pika.connection import ConnectionParameters
+
+        valid_vhost = "/neon_testing"
+        invalid_vhost = "/mock_vhost"
+        base_connection_kwargs = {"host": self.test_conf['server'],
+                                  "port": self.test_conf['port']}
+
+        valid_creds = PlainCredentials("test_user",
+                                       "test_password")
+        invalid_creds = PlainCredentials("test_user",
+                                         "invalid_password")
+
+        valid_connection = ConnectionParameters(**base_connection_kwargs,
+                                                virtual_host=valid_vhost,
+                                                credentials=valid_creds)
+        self.assertTrue(check_rmq_is_available(valid_connection))
+
+        invalid_bad_vhost = ConnectionParameters(**base_connection_kwargs,
+                                                 virtual_host=invalid_vhost,
+                                                 credentials=valid_creds)
+        with self.assertRaises(ProbableAccessDeniedError):
+            self.assertFalse(check_rmq_is_available(invalid_bad_vhost))
+
+        invalid_bad_creds = ConnectionParameters(**base_connection_kwargs,
+                                                 virtual_host=valid_vhost,
+                                                 credentials=invalid_creds)
+        with self.assertRaises(ProbableAuthenticationError):
+            self.assertFalse(check_rmq_is_available(invalid_bad_creds))
+
+        # If the calling service doesn't specify a `vhost`, allow it to start
+        # anyway (i.e. klat-observer)
+        invalid_default_vhost = ConnectionParameters(**base_connection_kwargs,
+                                                     virtual_host='/',
+                                                     credentials=valid_creds)
+        self.assertTrue(check_rmq_is_available(invalid_default_vhost))
 
 
 class TestConsumerUtils(unittest.TestCase):