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

http://git.mfd.msk/dashboard/issues?assignee_id=30

Browse source 
---
primary testing
This commit is contained in:
psemenkov 2022-09-12 13:11:12 +03:00
parent 10dd8bdd1d
commit 1bceabef85
2 changed files with 219 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
clearing-parent/account-service/pom.xml
View file

@ -36,6 +36,22 @@
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<!-- TEST -->
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>

203
clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java Normal file
View file

@ -0,0 +1,203 @@
package ru.spcex.clearing.account.service;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import ru.clearing.classes.statics.data.account.BankAccount;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgHazelcast;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.stream.StreamSupport;
import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
import static org.mockito.Mockito.doReturn;
class BankAccountServiceTest {
private static final int PARTITION = 0;
private static final String TOPIC = Consts.DESTINATION_BANK_ACCOUNT_NEW;
private MockConsumer<String, Integer> consumer;
private MockConsumer<String, Object> mockConsumer;
private List updates;
private CountryPopulationConsumer countryPopulationConsumer;
private Throwable pollException;
@BeforeEach
void setUp() {
mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
updates = new ArrayList<>();
countryPopulationConsumer = new CountryPopulationConsumer(consumer,
ex -> this.pollException = ex, updates::add);
}
@Test
public void bankAccountNew() {
//arrange
mockConsumer.schedulePollTask(() -> {
mockConsumer.rebalance(Collections.singletonList(new TopicPartition(TOPIC, PARTITION)));
mockConsumer.addRecord(new ConsumerRecord<>(TOPIC, PARTITION, 0, "key", "value"));
});
mockConsumer.schedulePollTask(() -> mockConsumer.wakeup());
HashMap<TopicPartition, Long> startOffsets = new HashMap<>();
TopicPartition tp = new TopicPartition(TOPIC, PARTITION);
startOffsets.put(tp, 0L);
mockConsumer.updateBeginningOffsets(startOffsets);
ImdgProvider imdgProvider = Mockito.mock(ImdgProvider.class);
Imdg<BankAccount> bankAccountMap = new ImdgHazelcast();
doReturn(bankAccountMap).when(imdgProvider).getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class);
BankAccountService bankAccountService = new BankAccountService(mockConsumer, imdgProvider);
// Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new LinkedHashMap<>();
//
// String topic = Consts.DESTINATION_BANK_ACCOUNT_NEW;
// ConsumerRecord<Integer, String> record1 = new ConsumerRecord<>(topic, 1, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 1, "value1");
// records.put(new TopicPartition(topic, 0), Arrays.asList(record1));
//doReturn(records).when(kafkaQueue).poll(Duration.of(10, ChronoUnit.SECONDS));
//act
bankAccountService.afterPropertiesSet();
//assert
}
@Test
void whenStartingByAssigningTopicPartition_thenExpectUpdatesAreConsumedCorrectly() {
// GIVEN
consumer.schedulePollTask(() -> consumer.addRecord(record(TOPIC, PARTITION, "Romania", 19_410_000)));
consumer.schedulePollTask(() -> countryPopulationConsumer.stop());
HashMap<TopicPartition, Long> startOffsets = new HashMap<>();
TopicPartition tp = new TopicPartition(TOPIC, PARTITION);
startOffsets.put(tp, 0L);
consumer.updateBeginningOffsets(startOffsets);
// WHEN
countryPopulationConsumer.startByAssigning(TOPIC, PARTITION);
// THEN
assertThat(updates).hasSize(1);
assertThat(consumer.closed()).isTrue();
}
@Test
void whenStartingBySubscribingToTopic_thenExpectUpdatesAreConsumedCorrectly() {
// GIVEN
consumer.schedulePollTask(() -> {
consumer.rebalance(Collections.singletonList(new TopicPartition(TOPIC, PARTITION)));
consumer.addRecord(record(TOPIC, PARTITION, "Romania", 20));
});
consumer.schedulePollTask(() -> countryPopulationConsumer.stop());
HashMap<TopicPartition, Long> startOffsets = new HashMap<>();
TopicPartition tp = new TopicPartition(TOPIC, 0);
startOffsets.put(tp, 0L);
consumer.updateBeginningOffsets(startOffsets);
// WHEN
countryPopulationConsumer.startBySubscribing(TOPIC);
// THEN
assertThat(updates).hasSize(1);
assertThat(consumer.closed()).isTrue();
}
class CountryPopulation {
private String country;
private Integer population;
// standard constructor, getters and setters
public CountryPopulation(String country, Integer population) {
this.country = country;
this.population = population;
}
public String getCountry() {
return country;
}
public void setCountry(String country) {
this.country = country;
}
public Integer getPopulation() {
return population;
}
public void setPopulation(Integer population) {
this.population = population;
}
}
public class CountryPopulationConsumer {
private Consumer<String, Integer> consumer;
private java.util.function.Consumer<Throwable> exceptionConsumer;
private java.util.function.Consumer<CountryPopulation> countryPopulationConsumer;
// standard constructor
public CountryPopulationConsumer(Consumer<String, Integer> consumer, java.util.function.Consumer<Throwable> exceptionConsumer, java.util.function.Consumer<CountryPopulation> countryPopulationConsumer) {
this.consumer = consumer;
this.exceptionConsumer = exceptionConsumer;
this.countryPopulationConsumer = countryPopulationConsumer;
}
void startBySubscribing(String topic) {
consume(() -> consumer.subscribe(Collections.singleton(topic)));
}
void startByAssigning(String topic, int partition) {
consume(() -> consumer.assign(Collections.singleton(new TopicPartition(topic, partition))));
}
private void consume(Runnable beforePollingTask) {
try {
beforePollingTask.run();
while (true) {
ConsumerRecords<String, Integer> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
StreamSupport.stream(records.spliterator(), false)
.map(record -> new CountryPopulation(record.key(), record.value()))
.forEach(countryPopulationConsumer);
consumer.commitSync();
}
} catch (WakeupException e) {
System.out.println("Shutting down...");
} catch (RuntimeException ex) {
exceptionConsumer.accept(ex);
} finally {
consumer.close();
}
}
public void stop() {
consumer.wakeup();
}
}
private ConsumerRecord<String, Integer> record(String topic, int partition, String country, int population) {
return new ConsumerRecord<>(topic, partition, 0, country, population);
}
}