juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/recorder/ApplicationTests.java

   1 package de.juplo.kafka.wordcount.recorder;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.consumer.ConsumerRecord;
   5 import org.junit.jupiter.api.Test;
   6 import org.springframework.beans.factory.annotation.Autowired;
   7 import org.springframework.boot.test.context.SpringBootTest;
   8 import org.springframework.boot.test.context.TestConfiguration;
   9 import org.springframework.boot.test.web.client.TestRestTemplate;
  10 import org.springframework.context.annotation.Bean;
  11 import org.springframework.http.MediaType;
  12 import org.springframework.kafka.annotation.KafkaListener;
  13 import org.springframework.kafka.test.context.EmbeddedKafka;
  14 import org.springframework.util.LinkedMultiValueMap;
  15 import org.springframework.util.MultiValueMap;
  16
  17 import static de.juplo.kafka.wordcount.recorder.ApplicationTests.PARTITIONS;
  18 import static de.juplo.kafka.wordcount.recorder.ApplicationTests.TOPIC_OUT;
  19 import static org.assertj.core.api.Assertions.assertThat;
  20
  21
  22 @SpringBootTest(
  23                 properties = {
  24                                 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  25                                 "juplo.wordcount.recorder.bootstrap-server=${spring.embedded.kafka.brokers}",
  26                                 "juplo.wordcount.recorder.topic=" + TOPIC_OUT })
  27 @EmbeddedKafka(topics = { TOPIC_OUT }, partitions = PARTITIONS)
  28 @Slf4j
  29 class ApplicationTests
  30 {
  31         public final static String TOPIC_OUT = "out";
  32         static final int PARTITIONS = 2;
  33
  34         @Autowired
  35         private TestRestTemplate restTemplate;
  36
  37         @Test
  38         void contextLoads()
  39         {
  40         }
  41
  42         @Test
  43         void userEventsAreSent()
  44         {
  45                 assertThat(restTemplate.postForObject("peter", "Hällö Wählt?*$@¢!", RecordingResult.class)).contains("Hello, World");
  46                 client
  47                                 .post()
  48                                 .uri("peter")
  49                                 .contentType(MediaType.TEXT_PLAIN)
  50                                 .bodyValue("Hallö Wält*&%€!")
  51                                 .
  52         }
  53
  54         static class Consumer
  55         {
  56                 final MultiValueMap<String, String> received = new LinkedMultiValueMap<>();
  57
  58                 @KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
  59                 public void receive(ConsumerRecord<String, String> record)
  60                 {
  61                         log.debug("Received message: {}", record);
  62                         received.add(record.key(), record.value());
  63                 }
  64         }
  65
  66         @TestConfiguration
  67         static class Configuration
  68         {
  69                 @Bean
  70                 Consumer consumer()
  71                 {
  72                         return new Consumer();
  73                 }
  74         }
  75 }