Issue #7: Send and expect requestId in record body, not in record key (#8)

2 files changed, 50 insertions, 16 deletions

diff --git a/src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt b/src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt
index 14bdd5d..3ec9757 100644
--- a/src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt
+++ b/src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt
@@ -97,9 +97,9 @@ class KafkaMtbFileSenderTest {
         val captor = argumentCaptor<String>()
         verify(kafkaTemplate, times(1)).send(anyString(), captor.capture(), captor.capture())
         assertThat(captor.firstValue).isNotNull
-        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\", \"eid\": \"1\", \"requestId\": \"TestID\"}")
+        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\", \"eid\": \"1\"}")
         assertThat(captor.secondValue).isNotNull
-        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(mtbFile(Consent.Status.ACTIVE)))
+        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(kafkaRecordData("TestID", Consent.Status.ACTIVE)))
     }
 
     @Test
@@ -113,9 +113,9 @@ class KafkaMtbFileSenderTest {
         val captor = argumentCaptor<String>()
         verify(kafkaTemplate, times(1)).send(anyString(), captor.capture(), captor.capture())
         assertThat(captor.firstValue).isNotNull
-        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\", \"requestId\": \"TestID\"}")
+        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\"}")
         assertThat(captor.secondValue).isNotNull
-        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(mtbFile(Consent.Status.REJECTED)))
+        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(kafkaRecordData("TestID", Consent.Status.REJECTED)))
     }
 
     companion object {
@@ -154,6 +154,10 @@ class KafkaMtbFileSenderTest {
             }.build()
         }
 
+        fun kafkaRecordData(requestId: String, consentStatus: Consent.Status): KafkaMtbFileSender.Data {
+            return KafkaMtbFileSender.Data(requestId, mtbFile(consentStatus))
+        }
+
         data class TestData(val requestStatus: RequestStatus, val exception: Throwable? = null)
 
         @JvmStatic
diff --git a/src/test/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessorTest.kt b/src/test/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessorTest.kt
index 6d83146..95bf41b 100644
--- a/src/test/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessorTest.kt
+++ b/src/test/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessorTest.kt
@@ -46,7 +46,7 @@ class KafkaResponseProcessorTest {
     private lateinit var kafkaResponseProcessor: KafkaResponseProcessor
 
     private fun createKafkaRecord(
-        requestId: String? = null,
+        requestId: String,
         statusCode: Int = 200,
         statusBody: Map<String, Any>? = mapOf()
     ): ConsumerRecord<String, String> {
@@ -54,15 +54,11 @@ class KafkaResponseProcessorTest {
             "test-topic",
             0,
             0,
-            if (requestId == null) {
-                null
-            } else {
-                this.objectMapper.writeValueAsString(KafkaResponseProcessor.ResponseKey(requestId))
-            },
+            null,
             if (statusBody == null) {
                 ""
             } else {
-                this.objectMapper.writeValueAsString(KafkaResponseProcessor.ResponseBody(statusCode, statusBody))
+                this.objectMapper.writeValueAsString(KafkaResponseProcessor.ResponseBody(requestId, statusCode, statusBody))
             }
         )
     }
@@ -78,17 +74,51 @@ class KafkaResponseProcessorTest {
     }
 
     @Test
-    fun shouldNotProcessRecordsWithoutValidKey() {
-        this.kafkaResponseProcessor.onMessage(createKafkaRecord(null, 200))
+    fun shouldNotProcessRecordsWithoutRequestIdInBody() {
+        val record = ConsumerRecord<String, String>(
+            "test-topic",
+            0,
+            0,
+            null,
+            """
+                {
+                    "statusCode": 200,
+                    "statusBody": {}
+                }
+            """.trimIndent()
+        )
+
+        this.kafkaResponseProcessor.onMessage(record)
+
+        verify(eventPublisher, never()).publishEvent(any<ResponseEvent>())
+    }
+
+    @Test
+    fun shouldProcessRecordsWithAliasNames() {
+        val record = ConsumerRecord<String, String>(
+            "test-topic",
+            0,
+            0,
+            null,
+            """
+                {
+                    "request_id": "test0123456789",
+                    "status_code": 200,
+                    "status_body": {}
+                }
+            """.trimIndent()
+        )
 
-        verify(eventPublisher, never()).publishEvent(any())
+        this.kafkaResponseProcessor.onMessage(record)
+
+        verify(eventPublisher, times(1)).publishEvent(any<ResponseEvent>())
     }
 
     @Test
-    fun shouldNotProcessRecordsWithoutValidBody() {
+    fun shouldNotProcessRecordsWithoutValidStatusBody() {
         this.kafkaResponseProcessor.onMessage(createKafkaRecord(requestId = "TestID1234", statusBody = null))
 
-        verify(eventPublisher, never()).publishEvent(any())
+        verify(eventPublisher, never()).publishEvent(any<ResponseEvent>())
     }
 
     @ParameterizedTest