WIP:KafkaChatHomeTest

diff --git a/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeTest.java b/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeTest.java
index 1298122..637f145 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeTest.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeTest.java
@@ -2,30 +2,49 @@ package de.juplo.kafka.chat.backend.persistence.kafka;
 
 import de.juplo.kafka.chat.backend.ChatBackendProperties;
 import de.juplo.kafka.chat.backend.domain.ChatHomeWithShardsTestBase;
-import de.juplo.kafka.chat.backend.persistence.kafka.messages.AbstractMessageTo;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
+import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.PropertySource;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import java.time.Clock;
-import java.time.ZoneId;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
+import static de.juplo.kafka.chat.backend.domain.ChatHomeWithShardsTestBase.NUM_SHARDS;
 import static de.juplo.kafka.chat.backend.persistence.kafka.KafkaChatHomeTest.TOPIC;
 
 
+@SpringBootTest(
+    classes = {
+        KafkaChatHomeTest.KafkaChatHomeTestConfiguration.class,
+        KafkaServicesConfiguration.class,
+        KafkaAutoConfiguration.class,
+        TaskExecutionAutoConfiguration.class,
+    },
+    properties = {
+    "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.chatroom-channel-topic=" + TOPIC,
+    "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
+})
 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
 @Slf4j
 public class KafkaChatHomeTest extends ChatHomeWithShardsTestBase
@@ -36,36 +55,17 @@ public class KafkaChatHomeTest extends ChatHomeWithShardsTestBase
 
 
   @TestConfiguration
-  static class Configuration
+  @EnableConfigurationProperties(ChatBackendProperties.class)
+  static class KafkaChatHomeTestConfiguration
   {
     @Bean
-    KafkaChatHome chatHome(ChatRoomChannel chatRoomChannel)
+    Clock clock()
     {
-      return new KafkaChatHome(numShards(), chatRoomChannel);
+      return Clock.systemDefaultZone();
     }
+  }
 
-    @Bean
-    ChatRoomChannel chatRoomChannel(
-        Producer<String, AbstractMessageTo> chatRoomChannelProducer,
-        Consumer<String, AbstractMessageTo> chatRoomChannelConsumer,
-        ZoneId zoneId,
-        Clock clock)
-    {
-      return new ChatRoomChannel(
-          TOPIC,
-          chatRoomChannelProducer,
-          chatRoomChannelConsumer,
-          zoneId,
-          numShards(),
-          8,
-          clock);
-    }
 
-    Integer numShards()
-    {
-      return 10;
-    }
-  }
   @BeforeAll
   public static void sendAndLoadStoredData(
       @Autowired KafkaTemplate<String, String> messageTemplate,