From 6a6f5260cfee2dc84feaf2a2a89ca4bec166b908 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sat, 3 Feb 2024 08:29:06 +0100
Subject: [PATCH] WIP

---
 .../chat/backend/AbstractHandoverIT.java      | 13 ++++
 .../kafka/chat/backend/KafkaHandoverIT.java   | 71 +++++++++++++++++++
 2 files changed, 84 insertions(+)
 create mode 100644 src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
 create mode 100644 src/test/java/de/juplo/kafka/chat/backend/KafkaHandoverIT.java

diff --git a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
new file mode 100644
index 00000000..0b8d3101
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@ -0,0 +1,13 @@
+package de.juplo.kafka.chat.backend;
+
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+
+@Testcontainers
+public abstract class AbstractHandoverIT
+{
+  KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
+    .withKraft();
+kafka.start();
+  var bootstrapServers = kafka.getBootstrapServers();
+}
diff --git a/src/test/java/de/juplo/kafka/chat/backend/KafkaHandoverIT.java b/src/test/java/de/juplo/kafka/chat/backend/KafkaHandoverIT.java
new file mode 100644
index 00000000..b126e091
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/chat/backend/KafkaHandoverIT.java
@@ -0,0 +1,71 @@
+package de.juplo.kafka.chat.backend;
+
+import de.juplo.kafka.chat.backend.implementation.kafka.ConsumerTaskRunner;
+import de.juplo.kafka.chat.backend.implementation.kafka.KafkaServicesApplicationRunner;
+import de.juplo.kafka.chat.backend.implementation.kafka.KafkaTestUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Import;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+
+import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.DATA_TOPIC;
+import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.INFO_TOPIC;
+
+
+@SpringBootTest(
+    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+    properties = {
+        "spring.main.allow-bean-definition-overriding=true",
+        "chat.backend.services=kafka",
+        "chat.backend.kafka.client-id-PREFIX=TEST",
+        "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
+        "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
+        "chat.backend.kafka.info-channel-topic=" + INFO_TOPIC,
+        "chat.backend.kafka.data-channel-topic=" + DATA_TOPIC,
+        "chat.backend.kafka.num-partitions=10",
+        })
+@EmbeddedKafka(
+    topics = { INFO_TOPIC, DATA_TOPIC },
+    partitions = 10)
+@Slf4j
+class KafkaHandoverIT extends AbstractHandoverIT
+{
+  final static String INFO_TOPIC = "KAFKA_CONFIGURATION_IT_INFO_CHANNEL";
+  final static String DATA_TOPIC = "KAFKA_CONFIGURATION_IT_DATA_CHANNEL";
+
+  @MockBean
+  KafkaServicesApplicationRunner kafkaServicesApplicationRunner;
+
+  @BeforeAll
+  public static void sendAndLoadStoredData(
+      @Autowired KafkaTemplate<String, String> messageTemplate,
+      @Autowired ConsumerTaskRunner consumerTaskRunner)
+  {
+    KafkaTestUtils.sendAndLoadStoredData(
+        messageTemplate,
+        INFO_TOPIC,
+        DATA_TOPIC,
+        consumerTaskRunner);
+  }
+
+  @AfterAll
+  static void joinConsumerTasks(
+      @Autowired ConsumerTaskRunner consumerTaskRunner)
+      throws InterruptedException
+  {
+    KafkaTestUtils.joinConsumerTasks(consumerTaskRunner);
+  }
+
+
+  @TestConfiguration
+  @Import(KafkaTestUtils.KafkaTestConfiguration.class)
+  static class KafkaConfigurationITConfiguration
+  {
+  }
+}
-- 
2.20.1