Verifying consumed message

Author: harishkannarao

app/consumer/inbound_consumer.py

@@ -1,8 +1,12 @@
 import asyncio
+import json
 from _asyncio import Task
 
 import structlog
 from aio_pika.abc import AbstractIncomingMessage
+from fastapi.encoders import jsonable_encoder
+
+from app.model.response.sample import Sample
 from app.rabbit_mq.rabbit_mq_client import get_connection
 from app.config import settings
 
@@ -12,10 +16,15 @@ async def process_inbound_message_task(message: AbstractIncomingMessage):
     async with message.process():  # Automatically ACKs if no exception occurs
         payload_string: str = message.body.decode()
         logger.info(
-            f"Processed inbound message {payload_string}", payload=payload_string
+            f"Received inbound message {payload_string}", payload=payload_string
+        )
+        samples = [Sample(**item) for item in json.loads(payload_string)]
+        logger.info(
+            f"Processed inbound message {payload_string}", samples=jsonable_encoder(samples)
         )
 
 
+
 async def start_inbound_consumer() -> Task[str]:
     logger = structlog.get_logger()
     # Create a dedicated channel for this consumer group

tests_integration/consumer/test_inbound_consumer.py

@@ -8,6 +8,7 @@
 import pytest
 import structlog
 from assertpy import assert_that
+from deepdiff import DeepDiff
 from fastapi.encoders import jsonable_encoder
 from fastapi.testclient import TestClient
 from tenacity import Retrying, stop_after_delay, wait_fixed
@@ -41,9 +42,9 @@ async def test_inbound_consumer_handles_message_successfully(
         updated_datetime=datetime.datetime.now(datetime.UTC),
         version=1,
     )
-    samples: list[Sample] = [sample1, sample2]
+    input_samples: list[Sample] = [sample1, sample2]
 
-    await publish_to_inbound(samples)
+    await publish_to_inbound(input_samples)
 
     for attempt in Retrying(
         stop=stop_after_delay(5), wait=wait_fixed(0.5), reraise=True
@@ -59,6 +60,14 @@ async def test_inbound_consumer_handles_message_successfully(
                 )
             )
             assert_that(inbound_consumer_logs).is_length(1)
+            consumed_samples: list[dict[str, Any]] = inbound_consumer_logs[0]["samples"]
+            assert_that(
+                DeepDiff(
+                    consumed_samples,
+                    jsonable_encoder(input_samples),
+                    ignore_order=True,
+                )
+            ).is_empty()
 
 
 async def publish_to_inbound(samples: list[Sample]):