CCM-12616: add senderId to CloudEvent data

Author: lapenna-bjss

lambdas/mesh-poll/src/__tests__/test_client_lookup.py

@@ -4,10 +4,10 @@
 import json
 import pytest
 from unittest.mock import Mock, call
+from src.client_lookup import ClientLookup
 
 
 def setup_mocks():
-
     ssm = Mock()
 
     config = Mock()
@@ -19,7 +19,6 @@ def setup_mocks():
 
 
 def create_client_parameter(client_id, mailbox_id):
-
     return {
         "Name": f"/dl/test/mesh/clients/{client_id}",
         "Value": json.dumps({
@@ -35,7 +34,6 @@ class TestClientLookup:
 
     def test_load_valid_senders_single_page(self):
         """Test loading valid senders from SSM (single page)"""
-        from src.client_lookup import ClientLookup
 
         ssm, config, logger = setup_mocks()
 
@@ -60,7 +58,6 @@ def test_load_valid_senders_single_page(self):
 
     def test_load_valid_senders_multiple_pages(self):
         """Test loading valid senders from SSM with pagination"""
-        from src.client_lookup import ClientLookup
 
         ssm, config, logger = setup_mocks()
 
@@ -85,17 +82,14 @@ def test_load_valid_senders_multiple_pages(self):
         assert ssm.get_parameters_by_path.call_count == 2
         ssm.get_parameters_by_path.assert_has_calls([
             call(Path="/dl/test/mesh/clients/", WithDecryption=True),
-            call(Path="/dl/test/mesh/clients/",
-                WithDecryption=True, NextToken="token123")
-        ])
+            call(Path="/dl/test/mesh/clients/", WithDecryption=True, NextToken="token123")
+        ], any_order=False)
         assert client_lookup.is_valid_sender("MAILBOX_001")
         assert client_lookup.is_valid_sender("MAILBOX_002")
         assert client_lookup.is_valid_sender("MAILBOX_003")
 
     def test_is_valid_sender_case_insensitive(self):
         """Test that sender validation is case-insensitive"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
 
         ssm.get_parameters_by_path.return_value = {
@@ -113,8 +107,6 @@ def test_is_valid_sender_case_insensitive(self):
 
     def test_is_valid_sender_returns_false_for_empty_mailbox_id(self):
         """Test that empty mailbox IDs are rejected"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
 
         ssm.get_parameters_by_path.return_value = {
@@ -130,8 +122,6 @@ def test_is_valid_sender_returns_false_for_empty_mailbox_id(self):
 
     def test_load_valid_senders_handles_malformed_json(self):
         """Test that malformed JSON in parameters is handled gracefully"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
 
         ssm.get_parameters_by_path.return_value = {
@@ -149,12 +139,10 @@ def test_load_valid_senders_handles_malformed_json(self):
 
         assert client_lookup.is_valid_sender("MAILBOX_001")
         assert client_lookup.is_valid_sender("MAILBOX_003")
-        logger.warn.assert_called()
+        assert logger.warn.called
 
     def test_load_valid_senders_handles_missing_mailbox_id(self):
         """Test that parameters without meshMailboxSenderId are skipped"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
 
         ssm.get_parameters_by_path.return_value = {
@@ -179,8 +167,6 @@ def test_load_valid_senders_handles_missing_mailbox_id(self):
 
     def test_load_valid_senders_handles_empty_mailbox_id(self):
         """Test that empty meshMailboxSenderId values are skipped"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
 
         ssm.get_parameters_by_path.return_value = {
@@ -206,8 +192,6 @@ def test_load_valid_senders_handles_empty_mailbox_id(self):
 
     def test_load_valid_senders_with_trailing_slash_in_path(self):
         """Test that paths with trailing slashes are handled correctly"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
         config.ssm_clients_parameter_path = "/dl/test/mesh/clients/"  # Trailing slash
 
@@ -227,8 +211,6 @@ def test_load_valid_senders_with_trailing_slash_in_path(self):
 
     def test_load_valid_senders_handles_empty_response(self):
         """Test that empty SSM response is handled correctly"""
-        from src.client_lookup import ClientLookup
-
         ssm, config, logger = setup_mocks()
 
         ssm.get_parameters_by_path.return_value = {

lambdas/mesh-poll/src/__tests__/test_processor.py

@@ -5,6 +5,7 @@
 import pytest
 from unittest.mock import Mock, call, patch
 from mesh_client import MeshClient
+from src.processor import MeshMessageProcessor
 
 
 def setup_mocks():
@@ -53,22 +54,20 @@ def setup_message_data(test_id="0"):
 
 
 def get_remaining_time_in_millis():
-    return 1000000
+    return 1000
 
 
 def get_remaining_time_in_millis_near_timeout():
     return 100
 
 
+@patch('src.processor.os.getenv')
+@patch('src.processor.EventPublisher')
 class TestMeshMessageProcessor:
     """Test suite for MeshMessageProcessor"""
 
-    @patch('src.processor.os.getenv')
-    @patch('src.processor.EventPublisher')
     def test_process_messages_iterates_through_inbox(self, mock_event_publisher_class, mock_getenv):
         """Test that processor iterates through all messages in MESH inbox"""
-        from src.processor import MeshMessageProcessor
-
         (config, client_lookup, mesh_client, log, polling_metric) = setup_mocks()
         message1 = setup_message_data("1")
         message2 = setup_message_data("2")
@@ -96,19 +95,13 @@ def test_process_messages_iterates_through_inbox(self, mock_event_publisher_clas
 
         mesh_client.handshake.assert_called_once()
         assert mesh_client.iterate_all_messages.call_count == 2
-        # Messages are acknowledged only on auth errors, not on success
         polling_metric.record.assert_called_once()
 
-    @patch('src.processor.os.getenv')
-    @patch('src.processor.EventPublisher')
     def test_process_messages_stops_near_timeout(self, mock_event_publisher_class, mock_getenv):
         """Test that processor stops processing when near timeout"""
-        from src.processor import MeshMessageProcessor
-
         (config, client_lookup, mesh_client, log, polling_metric) = setup_mocks()
         message1 = setup_message_data("1")
 
-        # Mock environment variables
         mock_getenv.side_effect = lambda key, default='': {
             'EVENT_PUBLISHER_EVENT_BUS_ARN': 'arn:aws:events:eu-west-2:123456789012:event-bus/test',
             'EVENT_PUBLISHER_DLQ_URL': 'https://sqs.eu-west-2.amazonaws.com/123456789012/test-dlq'
@@ -127,20 +120,14 @@ def test_process_messages_stops_near_timeout(self, mock_event_publisher_class, m
 
         processor.process_messages()
 
-        # Should not process messages when near timeout
         client_lookup.is_valid_sender.assert_not_called()
         polling_metric.record.assert_called_once()
 
-    @patch('src.processor.os.getenv')
-    @patch('src.processor.EventPublisher')
     def test_process_message_with_valid_sender(self, mock_event_publisher_class, mock_getenv):
         """Test processing a single message from valid sender"""
-        from src.processor import MeshMessageProcessor
-
         (config, client_lookup, mesh_client, log, polling_metric) = setup_mocks()
         message = setup_message_data("1")
 
-        # Mock environment variables
         mock_getenv.side_effect = lambda key, default='': {
             'EVENT_PUBLISHER_EVENT_BUS_ARN': 'arn:aws:events:eu-west-2:123456789012:event-bus/test',
             'EVENT_PUBLISHER_DLQ_URL': 'https://sqs.eu-west-2.amazonaws.com/123456789012/test-dlq'
@@ -168,16 +155,11 @@ def test_process_message_with_valid_sender(self, mock_event_publisher_class, moc
         mock_event_publisher.send_events.assert_called_once()
         message.acknowledge.assert_not_called()  # Only acknowledged on auth error
 
-    @patch('src.processor.os.getenv')
-    @patch('src.processor.EventPublisher')
     def test_process_message_with_unknown_sender(self, mock_event_publisher_class, mock_getenv):
         """Test that messages from unknown senders are rejected silently"""
-        from src.processor import MeshMessageProcessor
-
         (config, client_lookup, mesh_client, log, polling_metric) = setup_mocks()
         message = setup_message_data("1")
 
-        # Mock environment variables
         mock_getenv.side_effect = lambda key, default='': {
             'EVENT_PUBLISHER_EVENT_BUS_ARN': 'arn:aws:events:eu-west-2:123456789012:event-bus/test',
             'EVENT_PUBLISHER_DLQ_URL': 'https://sqs.eu-west-2.amazonaws.com/123456789012/test-dlq'
@@ -197,22 +179,16 @@ def test_process_message_with_unknown_sender(self, mock_event_publisher_class, m
 
         processor.process_message(message)
 
-        # Message should be acknowledged (removed from inbox)
         client_lookup.is_valid_sender.assert_called_once_with(message.sender)
         message.acknowledge.assert_called_once()
 
-    @patch('src.processor.os.getenv')
-    @patch('src.processor.EventPublisher')
     def test_process_messages_across_multiple_iterations(self, mock_event_publisher_class, mock_getenv):
         """Test that processor continues polling until no messages remain"""
-        from src.processor import MeshMessageProcessor
-
         (config, client_lookup, mesh_client, log, polling_metric) = setup_mocks()
         message1 = setup_message_data("1")
         message2 = setup_message_data("2")
         message3 = setup_message_data("3")
 
-        # Mock environment variables
         mock_getenv.side_effect = lambda key, default='': {
             'EVENT_PUBLISHER_EVENT_BUS_ARN': 'arn:aws:events:eu-west-2:123456789012:event-bus/test',
             'EVENT_PUBLISHER_DLQ_URL': 'https://sqs.eu-west-2.amazonaws.com/123456789012/test-dlq'
@@ -227,7 +203,6 @@ def test_process_messages_across_multiple_iterations(self, mock_event_publisher_
             polling_metric=polling_metric
         )
 
-        # Messages across multiple iterations
         mesh_client.iterate_all_messages.side_effect = [
             [message1, message2],  # First iteration
             [message3],            # Second iteration

lambdas/mesh-poll/src/client_lookup.py

@@ -4,14 +4,15 @@
 
 class ClientLookup:
     """
-    Lightweight client lookup for basic sender validation
+    Lightweight client lookup for basic sender validation and client ID extraction
     """
 
     def __init__(self, ssm, config, logger):
         self.__ssm = ssm
         self.__config = config
         self.__logger = logger
         self.__valid_senders = set()
+        self.__mailbox_to_client = {}
         self.load_valid_senders()
 
     def is_valid_sender(self, mailbox_id):
@@ -23,27 +24,39 @@ def is_valid_sender(self, mailbox_id):
 
         return mailbox_id.upper() in self.__valid_senders
 
+    def get_client_id(self, mailbox_id):
+        """
+        Get the client ID for a given MESH mailbox ID
+        """
+        if not mailbox_id:
+            return None
+
+        return self.__mailbox_to_client.get(mailbox_id.upper())
+
     def load_valid_senders(self):
         """
-        Loads just the mailbox IDs of valid senders into memory
+        Loads mailbox IDs and their corresponding client IDs into memory
         """
         mailbox_ids = set()
+        mailbox_to_client = {}
         next_token = ""
         page_number = 0
 
         while next_token or page_number < 1:
-            (page, token) = self.__get_page(next_token)
-            mailbox_ids.update(page)
+            (page_mailbox_ids, page_mapping, token) = self.__get_page(next_token)
+            mailbox_ids.update(page_mailbox_ids)
+            mailbox_to_client.update(page_mapping)
             next_token = token
             page_number += 1
 
         self.__valid_senders = mailbox_ids
+        self.__mailbox_to_client = mailbox_to_client
         self.__logger.debug(
             f"Loaded {len(self.__valid_senders)} valid sender mailbox IDs")
 
     def __get_page(self, next_token=""):
         """
-        Loads a page of client data and extracts just mailbox IDs
+        Loads a page of client data and extracts mailbox IDs and client IDs
         """
         if len(next_token) == 0:
             response = self.__ssm.get_parameters_by_path(
@@ -58,15 +71,19 @@ def __get_page(self, next_token=""):
             )
 
         mailbox_ids = set()
+        mailbox_to_client = {}
 
         if "Parameters" in response:
             for parameter in response["Parameters"]:
                 mailbox_id = self.__extract_mailbox_id(parameter)
-                if mailbox_id:
-                    mailbox_ids.add(mailbox_id.upper())
+                client_id = self.__extract_client_id(parameter)
+                if mailbox_id and client_id:
+                    mailbox_id_upper = mailbox_id.upper()
+                    mailbox_ids.add(mailbox_id_upper)
+                    mailbox_to_client[mailbox_id_upper] = client_id
 
         new_next_token = response.get("NextToken", "")
-        return (mailbox_ids, new_next_token)
+        return (mailbox_ids, mailbox_to_client, new_next_token)
 
     def __extract_mailbox_id(self, parameter):
         """
@@ -83,3 +100,19 @@ def __extract_mailbox_id(self, parameter):
                 f"Failed to parse mailbox ID from parameter {parameter['Name']}")
             self.__logger.error(format_exception(exception))
             return None
+
+    def __extract_client_id(self, parameter):
+        """
+        Extract just the client ID from a client parameter
+        """
+        if "Value" not in parameter:
+            return None
+
+        try:
+            client_config = json.loads(parameter["Value"])
+            return client_config.get("clientId", "")
+        except (ValueError, AttributeError) as exception:
+            self.__logger.warn(
+                f"Failed to parse client ID from parameter {parameter['Name']}")
+            self.__logger.error(format_exception(exception))
+            return None