1 files changed, 42 insertions, 4 deletions
diff --git a/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt b/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
index b4e4b92..d37c251 100644
--- a/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
+++ b/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
@@ -20,6 +20,7 @@
 package dev.dnpm.etl.processor.config
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import dev.dnpm.etl.processor.input.KafkaInputListener
 import dev.dnpm.etl.processor.monitoring.RequestRepository
 import dev.dnpm.etl.processor.output.KafkaMtbFileSender
 import dev.dnpm.etl.processor.output.RestMtbFileSender
@@ -78,8 +79,8 @@ class AppConfigurationTest {
     @TestPropertySource(
         properties = [
             "app.kafka.servers=localhost:9092",
-            "app.kafka.topic=test",
-            "app.kafka.response-topic=test-response",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
             "app.kafka.group-id=test"
         ]
     )
@@ -99,8 +100,8 @@ class AppConfigurationTest {
         properties = [
             "app.rest.uri=http://localhost:9000",
             "app.kafka.servers=localhost:9092",
-            "app.kafka.topic=test",
-            "app.kafka.response-topic=test-response",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
             "app.kafka.group-id=test"
         ]
     )
@@ -117,6 +118,43 @@ class AppConfigurationTest {
     @Nested
     @TestPropertySource(
         properties = [
+            "app.kafka.servers=localhost:9092",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
+            "app.kafka.group-id=test"
+        ]
+    )
+    inner class AppConfigurationWithoutKafkaInputTest(private val context: ApplicationContext) {
+
+        @Test
+        fun shouldNotUseKafkaInputListener() {
+            assertThrows<NoSuchBeanDefinitionException> { context.getBean(KafkaInputListener::class.java) }
+        }
+
+    }
+
+    @Nested
+    @TestPropertySource(
+        properties = [
+            "app.kafka.servers=localhost:9092",
+            "app.kafka.input-topic=test_input",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
+            "app.kafka.group-id=test"
+        ]
+    )
+    inner class AppConfigurationUsingKafkaInputTest(private val context: ApplicationContext) {
+
+        @Test
+        fun shouldUseKafkaInputListener() {
+            assertThat(context.getBean(KafkaInputListener::class.java)).isNotNull
+        }
+
+    }
+
+    @Nested
+    @TestPropertySource(
+        properties = [
             "app.transformations[0].path=consent.status",
             "app.transformations[0].from=rejected",
             "app.transformations[0].to=accept",