fix wildcard issue, add integration test and CI/CD

Signed-off-by: Lance-Drane <Lance-Drane@users.noreply.github.com>

Author: Lance-Drane

.github/workflows/ci.yml

@@ -36,15 +36,54 @@ jobs:
 
   # Run unit tests only on Windows/MacOS, we can run the full test suite on Linux
   test-unit:
-    name: Unit tests
+    name: Unit tests for Windows, MacOS
     strategy:
       matrix:
         python-version: ["3.10", "3.11", "3.12", "3.13"]
         os:
-          - ubuntu-latest
           - macos-latest
           - windows-latest
     runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup UV
+        uses: astral-sh/setup-uv@v7
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          uv sync --locked --all-extras --all-groups
+      - name: Run tests
+        run: uv run pytest tests/unit/ --cov=src/intersect_sdk_common/ --cov-report=html:reports/htmlcov/ --cov-report=xml:reports/coverage_report.xml --junitxml=reports/junit.xml
+
+  test-suite:
+    name: Full test suite with RabbitMQ and MinIO on Linux
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
+    runs-on: ubuntu-latest
+    services:
+      rabbitmq:
+        image: "bitnamilegacy/rabbitmq:4.1"
+        env:
+          # space-delimited list of plugins
+          RABBITMQ_PLUGINS: "rabbitmq_mqtt"
+          RABBITMQ_USERNAME: "intersect_username"
+          RABBITMQ_PASSWORD: "intersect_password"
+          # misleading env name, this is needed to set "loopback_users.$USERNAME = false"
+          RABBITMQ_MANAGEMENT_ALLOW_WEB_ACCESS: "yes"
+        ports:
+          - "1883:1883" # MQTT port
+          - "5672:5672" # AMQP port
+      minio:
+        image: "bitnamilegacy/minio:2024.6.4"
+        env:
+          MINIO_ROOT_USER: AKIAIOSFODNN7EXAMPLE
+          MINIO_ROOT_PASSWORD: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
+          MINIO_SKIP_CLIENT: "yes"
+        ports:
+          - "9000:9000" # main MINIO API port
+          - "9001:9001" # web UI
     steps:
       - uses: actions/checkout@v4
       - name: Setup UV

src/intersect_sdk_common/control_plane/brokers/amqp_client.py

@@ -23,14 +23,12 @@
 from .broker_client import BrokerClient
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
-
     from pika.channel import Channel
     from pika.frame import Frame
     from pika.spec import Basic, BasicProperties
 
+    from ..control_plane_manager import ControlPlaneManager
     from ..definitions import MessageCallback
-    from ..topic_handler import TopicHandler
 
 
 _AMQP_MAX_RETRIES = 10
@@ -95,8 +93,7 @@ def __init__(
         port: int,
         username: str,
         password: str,
-        topics_to_handlers: Callable[[], dict[str, TopicHandler]],
-        find_wildcard_handler: Callable[[str], TopicHandler | None],
+        control_plane_manager: ControlPlaneManager,
         is_root: bool,
     ) -> None:
         """The default constructor.
@@ -106,8 +103,7 @@ def __init__(
             port: port number of AMQP broker
             username: username credentials for AMQP broker
             password: password credentials for AMQP broker
-            topics_to_handlers: callback function which gets the topic to handler map from the channel manager
-            find_wildcard_handler: callback function which gets the wildcard topic handler from the channel manager
+            control_plane_manager: reference to the ControlPlaneManager instance, remember to ONLY use functions which do not mutate state
             is_root: Whether or not the client can configure exchanges and queues themselves (core services), or if this must be delegated to a Core Service (SDK Clients/Services)
         """
         self._connection_params = pika.ConnectionParameters(
@@ -132,11 +128,11 @@ def __init__(
         self._thread: threading.Thread | None = None
 
         # Callback to the topics_to_handler list inside of
-        self._topics_to_handlers = topics_to_handlers
-        self._find_wildcard_handler = find_wildcard_handler
+        self._control_plane_manager = control_plane_manager
 
         # mapping of topics to callables which can unsubscribe from the topic
         self._topics_to_consumer_tags: dict[str, _ConsumerTagInfo] = {}
+        """NOTE: separators for topics in this object is '.' , not '/' """
         self._consumer_tags_to_threads: dict[str, threading.Thread] = {}
 
         self._should_disconnect = False
@@ -390,7 +386,7 @@ def _on_output_channel_open(self, channel: Channel) -> None:
     def _on_input_channel_open(self, channel: Channel) -> None:
         channel_num = 1
         self._channel_in = channel
-        # consumer channel flag can be set immediately
+        # consumer channel flag can be set immediately (NOTE: this means that code which simultaneously creates Clients and Services will need to manually wait, to ensure that each application has made the basic_consume call to the broker before messages are published)
         self._channel_flags.set_nth_flag(channel_num)
         cb_1 = functools.partial(self._on_channel_closed, channel_num=channel_num)
         self._channel_in.add_on_close_callback(cb_1)
@@ -403,17 +399,7 @@ def _on_input_channel_open(self, channel: Channel) -> None:
                 callback=cb_2,
             )
         else:
-            for topic, topic_handler in self._topics_to_handlers().items():
-                amqp_topic = _hierarchy_2_amqp(topic)
-                cb = functools.partial(
-                    self._setup_basic_qos,
-                    None,
-                    channel=channel,
-                    topic=amqp_topic,
-                    persist=topic_handler.topic_persist,
-                    queue_name=topic_handler.queue_name,
-                )
-                self._connection.ioloop.add_callback_threadsafe(cb)
+            self._create_all_queues(channel)
 
     def _on_exchange_declareok(self, _unused_frame: Frame, channel: Channel) -> None:
         """Create a queue on the broker (called from AMQP).
@@ -425,7 +411,17 @@ def _on_exchange_declareok(self, _unused_frame: Frame, channel: Channel) -> None
             _unused_frame: response from declaring the exchange on the broker (irrelevant).
             channel: The Channel being instantiated.
         """
-        for topic, topic_handler in self._topics_to_handlers().items():
+        self._create_all_queues(channel)
+
+    def _create_all_queues(self, channel: Channel) -> None:
+        """Create queues for all subscription channels on the broker.
+
+        This should only be called once we've verified that the exchange exists.
+
+        Args:
+            channel: The Channel being instantiated.
+        """
+        for topic, topic_handler in self._control_plane_manager.get_all_subscription_channels():
             amqp_topic = _hierarchy_2_amqp(topic)
             cb = functools.partial(
                 self._create_queue,
@@ -456,6 +452,7 @@ def _create_queue(self, channel: Channel, topic: str, persist: bool, queue_name:
             queue=queue_name
             if persist
             else '',  # if we're transient, let the broker generate a name for us
+            passive=not self._is_root,
             durable=persist,
             exclusive=not persist,  # transient queues can be exclusive
             callback=cb,
@@ -597,17 +594,25 @@ def _consume_message(
             channel.basic_ack(basic_deliver.delivery_tag)
             return
 
-        tth_key = _amqp_2_hierarchy(basic_deliver.routing_key)
-        topic_handler = self._topics_to_handlers().get(tth_key)
+        amqp_routing_key_rep = basic_deliver.routing_key
+        manager_routing_key_rep = _amqp_2_hierarchy(basic_deliver.routing_key)
+        topic_handler = self._control_plane_manager.get_non_wildcard_subscription_channels().get(
+            manager_routing_key_rep
+        )
         if not topic_handler:
             # we may have included the topic in one of our wildcard handlers
-            topic_handler = self._find_wildcard_handler(tth_key)
+            wildcard_result = self._control_plane_manager.get_wildcard_topic_and_topic_handler(
+                manager_routing_key_rep
+            )
+            if wildcard_result:
+                manager_routing_key_rep, topic_handler = wildcard_result
+                amqp_routing_key_rep = _hierarchy_2_amqp(manager_routing_key_rep)
         if topic_handler:
-            consumer_tag_info = self._topics_to_consumer_tags.get(basic_deliver.routing_key)
+            consumer_tag_info = self._topics_to_consumer_tags.get(amqp_routing_key_rep)
             if not consumer_tag_info:
                 logger.error(
                     'Could not fetch consumer tag for topic %s, please inform an INTERSECT-SDK developer that you saw this message',
-                    tth_key,
+                    amqp_routing_key_rep,
                 )
                 return
             while not consumer_tag_info.consumer_tag_obtained():

src/intersect_sdk_common/control_plane/brokers/mqtt_client.py

@@ -22,12 +22,10 @@
 from .broker_client import BrokerClient
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
-
     from paho.mqtt.client import DisconnectFlags
     from paho.mqtt.reasoncodes import ReasonCode
 
-    from ..topic_handler import TopicHandler
+    from ..control_plane_manager import ControlPlaneManager
 
 
 _MQTT_MAX_RETRIES = 10
@@ -37,7 +35,8 @@
 # this will be a breaking change, so only add it when ready to break
 def _hierarchy_2_mqtt(hierarchy: str) -> str:
     """Take the hierarchy string format saved in the Service and map it to the MQTT topic format. Currently just covers wildcards."""
-    return hierarchy.replace('#', '+')
+    # return hierarchy.replace('#', '+')
+    return hierarchy.replace('*', '+')
 
 
 class MQTTClient(BrokerClient):
@@ -61,8 +60,7 @@ def __init__(
         port: int,
         username: str,
         password: str,
-        topics_to_handlers: Callable[[], dict[str, TopicHandler]],
-        find_wildcard_handler: Callable[[str], TopicHandler | None],
+        control_plane_manager: ControlPlaneManager,
         uid: str | None = None,
     ) -> None:
         """The default constructor.
@@ -72,8 +70,7 @@ def __init__(
             port: port number of MQTT broker
             username: username credentials for MQTT broker
             password: password credentials for MQTT broker
-            topics_to_handlers: callback function which gets the topic to handler map from the channel manager
-            find_wildcard_handler: callback function which gets the wildcard topic handler from the channel manager
+            control_plane_manager: reference to the ControlPlaneManager instance, remember to ONLY use functions which do not mutate state
             uid: A string representing the unique id to identify the client.
         """
         # Unique id for the MQTT broker to associate this client with
@@ -97,8 +94,7 @@ def __init__(
         self._connected_flag = threading.Event()
 
         # ConnectionManager callable state
-        self._topics_to_handlers = topics_to_handlers
-        self._find_wildcard_handler = find_wildcard_handler
+        self._control_plane_manager = control_plane_manager
 
         # MQTT v3.1.1 automatically downgrades a QOS which is too high (good), but MQTT v5 will terminate the connection (bad)
         # see https://github.com/rabbitmq/rabbitmq-server/discussions/11842
@@ -204,10 +200,17 @@ def _on_message(
           message: MQTT message
         """
         # NOTE: should not need to convert the MQTT topic to the protocol agnostic representation, as we only change the wildcard format (which won't show up as the message topic)
-        topic_handler = self._topics_to_handlers().get(message.topic)
+        inmem_topic = message.topic
+        topic_handler = self._control_plane_manager.get_non_wildcard_subscription_channels().get(
+            inmem_topic
+        )
         if not topic_handler:
             # we may have included the topic in one of our wildcard handlers
-            topic_handler = self._find_wildcard_handler(message.topic)
+            wildcard_result = self._control_plane_manager.get_wildcard_topic_and_topic_handler(
+                inmem_topic
+            )
+            if wildcard_result:
+                _, topic_handler = wildcard_result
         # Note that if we return prior to the callback, there will be no reply message
         if not topic_handler:
             logger.warning('Incompatible message topic %s, rejecting message', message.topic)
@@ -298,7 +301,7 @@ def _handle_connect(
                 self._max_supported_qos = properties.MaximumQoS
 
             self._connected_flag.set()
-            for topic, topic_handler in self._topics_to_handlers().items():
+            for topic, topic_handler in self._control_plane_manager.get_all_subscription_channels():
                 self.subscribe(topic, topic_handler.topic_persist)
         else:
             # This will generally suggest a misconfiguration

src/intersect_sdk_common/control_plane/control_plane_manager.py

@@ -1,7 +1,7 @@
 """Wrapper around interacting with the control plane, which may be multiple brokers."""
 
+import itertools
 import re
-from collections.abc import Callable
 
 from ..config import ControlPlaneConfig
 from ..exceptions import IntersectSetupError
@@ -23,8 +23,7 @@
 
 def _create_control_provider(
     config: ControlPlaneConfig,
-    topic_handler_callback: Callable[[], dict[str, TopicHandler]],
-    find_wildcard_handler: Callable[[str], TopicHandler | None],
+    control_plane_manager: 'ControlPlaneManager',
 ) -> BrokerClient:
     if config.protocol == 'amqp0.9.1':
         from .brokers.amqp_client import (  # noqa: PLC0415 (lazy load all AMQP modules)
@@ -36,8 +35,7 @@ def _create_control_provider(
             port=config.port or 5672,
             username=config.username,
             password=config.password,
-            topics_to_handlers=topic_handler_callback,
-            find_wildcard_handler=find_wildcard_handler,
+            control_plane_manager=control_plane_manager,
             is_root=config.is_root,
         )
 
@@ -49,8 +47,7 @@ def _create_control_provider(
         port=config.port or 1883,
         username=config.username,
         password=config.password,
-        topics_to_handlers=topic_handler_callback,
-        find_wildcard_handler=find_wildcard_handler,
+        control_plane_manager=control_plane_manager,
     )
 
 
@@ -68,10 +65,7 @@ def __init__(
         queue_name_generator should be a hardcoded value for Core Services, the SDK should provide its own function to generate queue names.
         """
         self._control_providers = [
-            _create_control_provider(
-                config, self.get_subscription_channels, self.get_wildcard_topic_handler
-            )
-            for config in control_configs
+            _create_control_provider(config, self) for config in control_configs
         ]
 
         # flag which indicates if we SHOULD be connected.
@@ -142,29 +136,47 @@ def remove_subscription_channel(self, channel: str) -> bool:
                 provider.unsubscribe(channel)
         return True
 
-    def get_subscription_channels(self) -> dict[str, TopicHandler]:
-        """Get the subscription channels.
+    def get_non_wildcard_subscription_channels(self) -> dict[str, TopicHandler]:
+        """Get primary subscription channels.
 
-        These channels cannot be wildcards, and incoming topics must match the channel topics exactly. Note that this function gets accessed as a callback from the direct broker implementations.
+        These channels cannot be wildcards, and incoming topics must match the channel topics exactly.
+
+        This function is safe to call from the broker clients, as it does not mutate state.
 
         Returns:
           the dictionary of topics to topic information
         """
         return self._topics_to_handlers
 
-    def get_wildcard_topic_handler(self, topic: str) -> TopicHandler | None:
+    def get_wildcard_topic_and_topic_handler(self, topic: str) -> tuple[str, TopicHandler] | None:
         """Get the wildcard topic handler for a given topic, if it exists.
 
         This is an inefficient lookup, so should only be used if we fail to find a non-wildcard match for an incoming topic. Note that this function gets accessed as a callback from the direct broker implementations.
 
+        This function is safe to call from the broker clients, as it does not mutate state.
+
+        Params:
+          topic: the topic for an incoming message (this will not have any wildcards in it)
+
         Returns:
-          the topic handler associated with the topic, if it exists; None otherwise
+          if a match found: the wildcard string configured application-side to match the topic parameter, and the topic handler associated with the topic
+          if no match found: None
         """
-        for topic_handler in self._wildcards.values():
+        for wildcard, topic_handler in self._wildcards.items():
             if topic_handler.does_topic_match(topic):
-                return topic_handler
+                return wildcard, topic_handler
         return None
 
+    def get_all_subscription_channels(self) -> itertools.chain[tuple[str, TopicHandler]]:
+        """Get all subscription channels, including wildcard channels.
+
+        This function is safe to call from the broker clients, as it does not mutate state.
+
+        Returns:
+            an iterator which yields tuples of (channel, topic handler) for all channels, including wildcards. Note that wildcard channels will be returned in no particular order, and may be interspersed with non-wildcard channels.
+        """
+        return itertools.chain(self._topics_to_handlers.items(), self._wildcards.items())
+
     def connect(self) -> None:
         """Connect to all configured brokers.
 

src/intersect_sdk_common/control_plane/topic_handler.py

@@ -46,8 +46,8 @@ def __init__(
                     # match any sequence of 0 or more words (make sure to allow for the separator)
                     regex_builder.append('[a-zA-Z0-9/-]*')
                 elif char == '/':
-                    # in-memory topics use '/' as the separator, but on the broker '.' is used
-                    regex_builder.append('\\.')
+                    # in-memory topics use '/' as the separator, it is the protocol handler's responsibility to convert topics to use '/' as the separator
+                    regex_builder.append('/')
                 else:
                     # ordinary character
                     regex_builder.append(char)