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 com.fasterxml.jackson.databind.ObjectMapper;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.clients.consumer.ConsumerRecord;
   6 import org.junit.jupiter.api.DisplayName;
   7 import org.junit.jupiter.api.Test;
   8 import org.springframework.beans.factory.annotation.Autowired;
   9 import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
  10 import org.springframework.boot.test.context.SpringBootTest;
  11 import org.springframework.boot.test.context.TestConfiguration;
  12 import org.springframework.context.annotation.Bean;
  13 import org.springframework.http.MediaType;
  14 import org.springframework.kafka.annotation.KafkaListener;
  15 import org.springframework.kafka.test.context.EmbeddedKafka;
  16 import org.springframework.test.web.servlet.MockMvc;
  17 import org.springframework.test.web.servlet.MvcResult;
  18 import org.springframework.util.LinkedMultiValueMap;
  19 import org.springframework.util.MultiValueMap;
  20
  21 import java.time.Duration;
  22
  23 import static de.juplo.kafka.wordcount.recorder.ApplicationTests.PARTITIONS;
  24 import static de.juplo.kafka.wordcount.recorder.ApplicationTests.TOPIC_OUT;
  25 import static org.assertj.core.api.Assertions.assertThat;
  26 import static org.awaitility.Awaitility.await;
  27 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
  28 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
  29 import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
  30 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
  31 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
  32
  33
  34 @SpringBootTest(
  35                 properties = {
  36                                 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  37                                 "juplo.wordcount.recorder.bootstrap-server=${spring.embedded.kafka.brokers}",
  38                                 "juplo.wordcount.recorder.topic=" + TOPIC_OUT })
  39 @AutoConfigureMockMvc
  40 @EmbeddedKafka(topics = { TOPIC_OUT }, partitions = PARTITIONS)
  41 @Slf4j
  42 class ApplicationTests
  43 {
  44         static final String TOPIC_OUT = "out";
  45         static final int PARTITIONS = 2;
  46         static final String USER = "päter";
  47         static final String SENTENCE = "Hall° Wält?¢*&%€!";
  48
  49
  50         @Autowired
  51         MockMvc mockMvc;
  52         @Autowired
  53         Consumer consumer;
  54         @Autowired
  55         ObjectMapper objectMapper;
  56
  57
  58         @Test
  59         @DisplayName("The application context loads")
  60         void contextLoads()
  61         {
  62         }
  63
  64         @Test
  65         @DisplayName("Posted messages are excepted and sent to Kafka")
  66         void userMessagesAreExceptedAndSentToKafka() throws Exception
  67         {
  68                 MvcResult result = mockMvc
  69                                 .perform(post("/{user}", USER)
  70                                                 .contentType(MediaType.TEXT_PLAIN)
  71                                                 .content(SENTENCE))
  72                                 .andReturn();
  73
  74                 mockMvc.perform(asyncDispatch(result))
  75                                 .andDo(print())
  76                                 .andExpect(status().isOk())
  77                                 .andExpect(jsonPath("$.username").value(USER))
  78                                 .andExpect(jsonPath("$.sentence").value(SENTENCE));
  79
  80                 await("Expexted converted data")
  81                                 .atMost(Duration.ofSeconds(10))
  82                                 .untilAsserted(() ->
  83                                 {
  84                                         assertThat(consumer.received.get(USER)).hasSize(1);
  85                                         RecordingResult recordingResult = objectMapper.readValue(
  86                                                         consumer.received.get(USER).get(0),
  87                                                         RecordingResult.class);
  88                                         assertThat(recordingResult.getUsername()).isEqualTo(USER);
  89                                         assertThat(recordingResult.getSentence()).isEqualTo(SENTENCE);
  90                                         assertThat(recordingResult.getTopic()).isEqualTo(TOPIC_OUT);
  91                                         assertThat(recordingResult.getPartition()).isBetween(0, PARTITIONS - 1);
  92                                         assertThat(recordingResult.getStatus()).isNull();
  93                                         assertThat(recordingResult.getError()).isNull();
  94                                 });
  95         }
  96
  97         static class Consumer
  98         {
  99                 final MultiValueMap<String, String> received = new LinkedMultiValueMap<>();
 100
 101                 @KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
 102                 public void receive(ConsumerRecord<String, String> record)
 103                 {
 104                         log.debug("Received message: {}", record);
 105                         received.add(record.key(), record.value());
 106                 }
 107         }
 108
 109         @TestConfiguration
 110         static class Configuration
 111         {
 112                 @Bean
 113                 Consumer consumer()
 114                 {
 115                         return new Consumer();
 116                 }
 117         }
 118 }