treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Delete test with EmbeddedKafka.

Browse source
This commit is contained in:
psemenkov 2023-01-30 11:38:32 +03:00
parent 8c4e5c3d40
commit 5d9a66eece
5 changed files with 16 additions and 241 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
clearing-parent/balance-service/pom.xml
View file

@ -14,7 +14,6 @@
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<external_libraries.kafka.version>3.0.1</external_libraries.kafka.version>
</properties>
<dependencies>
<dependency>
@ -57,23 +56,6 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<version>2.8.8</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.8.8</version>
</dependency>
</dependencies>
<build>
<resources>

25
clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/config/KafkaTestConfig.java
View file

@ -1,17 +1,24 @@
package ru.spcex.clearing.balance.config;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.spcex.clearing.platform.messaging.serialization.JsonSerializer;
@Configuration
public class KafkaTestConfig {
// @Bean
// public MockConsumer<String, Object> createTestConsumer() {
// return new MockConsumer<>(OffsetResetStrategy.EARLIEST);
// }
//
// @Bean
// public Producer<String, Object> createTestProducer() {
// return new MockProducer<>(true, new StringSerializer(), new JsonSerializer());
// }
@Bean
public MockConsumer<String, Object> createTestConsumer() {
return new MockConsumer<>(OffsetResetStrategy.EARLIEST);
}
@Bean
public Producer<String, Object> createTestProducer() {
return new MockProducer<>(true, new StringSerializer(), new JsonSerializer());
}
}

41
clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/config/KafkaTestConsumer.java
View file

@ -1,41 +0,0 @@
package ru.spcex.clearing.balance.config;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.CountDownLatch;
@Component
public class KafkaTestConsumer {
public static final String TOPIC_GALB = "launcher-GALB";
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaTestConsumer.class);
private final String TOPIC_NAME = "com.madadipouya.kafka.user";
private CountDownLatch latch = new CountDownLatch(1);
private String payload;
@KafkaListener(topics = TOPIC_GALB)
public void receive(ConsumerRecord<?, ?> consumerRecord) {
LOGGER.info("received payload='{}'", consumerRecord.toString());
payload = consumerRecord.toString();
latch.countDown();
}
public CountDownLatch getLatch() {
return latch;
}
public void resetLatch() {
latch = new CountDownLatch(1);
}
public String getPayload() {
return payload;
}
}

104
clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/service/EmbeddedKafkaTest.java
View file

@ -1,104 +0,0 @@
package ru.spcex.clearing.balance.service;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.*;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.ContainerTestUtils;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import ru.spcex.clearing.balance.config.KafkaTestConsumer;
import java.io.File;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static ru.spcex.clearing.balance.config.KafkaTestConsumer.TOPIC_GALB;
@EmbeddedKafka//(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@ExtendWith(SpringExtension.class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EmbeddedKafkaTest {
static {
Path path = Paths.get("src", "main", "resources");
String currentPath = path.toAbsolutePath().toString();
System.setProperty("spring.config.location", currentPath + File.separator);
}
@Autowired
public KafkaTemplate<String, String> template;
BlockingQueue<ConsumerRecord<String, String>> records;
KafkaMessageListenerContainer<String, String> container;
@Autowired
private EmbeddedKafkaBroker embeddedKafkaBroker;
@Autowired
private KafkaTestConsumer consumer;
@BeforeEach
void setup() {
consumer.resetLatch();
}
@BeforeAll
void setUp() {
Map<String, Object> configs = new HashMap<>(KafkaTestUtils.consumerProps("consumer", "false", embeddedKafkaBroker));
DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new StringDeserializer());
ContainerProperties containerProperties = new ContainerProperties(TOPIC_GALB);
container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
records = new LinkedBlockingQueue<>();
container.setupMessageListener((MessageListener<String, String>) records::add);
container.start();
ContainerTestUtils.waitForAssignment(container, embeddedKafkaBroker.getPartitionsPerTopic());
}
@AfterAll
void tearDown() {
container.stop();
}
@Test
public void kafkaSetup_withTopic_ensureSendMessageIsReceived() throws Exception {
// Arrange
Map<String, Object> configs = new HashMap<>(KafkaTestUtils.producerProps(embeddedKafkaBroker));
Producer<String, String> producer = new DefaultKafkaProducerFactory<>(configs, new StringSerializer(), new StringSerializer()).createProducer();
String data = "Sending with default template";
//Act
producer.send(new ProducerRecord<>(TOPIC_GALB, "my-aggregate-id", data));
// producer.flush();
// template.send(TOPIC, data);
// Assert
ConsumerRecord<String, String> singleRecord = records.poll(10, TimeUnit.SECONDS);
boolean messageConsumed = consumer.getLatch()
.await(30, TimeUnit.SECONDS);
assertTrue(messageConsumed);
assertThat(consumer.getPayload(), containsString(data));
// assertThat(singleRecord).isNotNull();
// assertThat(singleRecord.key()).isEqualTo("my-aggregate-id");
// assertThat(singleRecord.value()).isEqualTo("{\"event\":\"Test Event\"}");
}
}

69
clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/service/Sdf08ServiceKafkaTest.java
View file

@ -1,69 +0,0 @@
package ru.spcex.clearing.balance.service;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import ru.spcex.clearing.balance.config.KafkaTestConsumer;
import java.io.File;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static ru.spcex.clearing.balance.config.KafkaTestConsumer.TOPIC_GALB;
@EmbeddedKafka//(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@ExtendWith(SpringExtension.class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class Sdf08ServiceKafkaTest {
static {
Path path = Paths.get("src", "main", "resources");
String currentPath = path.toAbsolutePath().toString();
System.setProperty("spring.config.location", currentPath + File.separator);
}
@Autowired
public KafkaTemplate<String, String> template;
@Autowired
private EmbeddedKafkaBroker embeddedKafkaBroker;
@Autowired
private KafkaTestConsumer consumer;
@Test
public void newSDf08() throws Exception {
// Arrange
Map<String, Object> configs = new HashMap<>(KafkaTestUtils.producerProps(embeddedKafkaBroker));
Producer<String, String> producer = new DefaultKafkaProducerFactory<>(configs, new StringSerializer(), new StringSerializer()).createProducer();
String data = "Sending with default template";
//Act
producer.send(new ProducerRecord<>(TOPIC_GALB, "my-aggregate-id", data));
// producer.flush();
// template.send(TOPIC, data);
// Assert
boolean messageConsumed = consumer.getLatch()
.await(10, TimeUnit.SECONDS);
assertTrue(messageConsumed);
assertThat(consumer.getPayload(), containsString(data));
}
}