From f8271d3ee2909c6e03f710635d77b40f1f9a5500 Mon Sep 17 00:00:00 2001 From: psemenkov Date: Mon, 12 Sep 2022 13:11:12 +0300 Subject: [PATCH] http://git.mfd.msk/dashboard/issues?assignee_id=30 --- primary testing --- clearing-parent/account-service/pom.xml | 16 ++ .../service/BankAccountServiceTest.java | 203 ++++++++++++++++++ 2 files changed, 219 insertions(+) create mode 100644 clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java diff --git a/clearing-parent/account-service/pom.xml b/clearing-parent/account-service/pom.xml index 7585732cf..01f40da8d 100644 --- a/clearing-parent/account-service/pom.xml +++ b/clearing-parent/account-service/pom.xml @@ -36,6 +36,22 @@ com.fasterxml.jackson.core jackson-databind + + + org.mockito + mockito-core + + + org.junit.jupiter + junit-jupiter + test + + + org.assertj + assertj-core + test + + diff --git a/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java new file mode 100644 index 000000000..907a76f6e --- /dev/null +++ b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java @@ -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 consumer; + private MockConsumer 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 startOffsets = new HashMap<>(); + TopicPartition tp = new TopicPartition(TOPIC, PARTITION); + startOffsets.put(tp, 0L); + mockConsumer.updateBeginningOffsets(startOffsets); + + ImdgProvider imdgProvider = Mockito.mock(ImdgProvider.class); + Imdg bankAccountMap = new ImdgHazelcast(); + doReturn(bankAccountMap).when(imdgProvider).getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class); + + BankAccountService bankAccountService = new BankAccountService(mockConsumer, imdgProvider); +// Map>> records = new LinkedHashMap<>(); +// +// String topic = Consts.DESTINATION_BANK_ACCOUNT_NEW; +// ConsumerRecord 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 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 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 consumer; + private java.util.function.Consumer exceptionConsumer; + private java.util.function.Consumer countryPopulationConsumer; + + // standard constructor + + + public CountryPopulationConsumer(Consumer consumer, java.util.function.Consumer exceptionConsumer, java.util.function.Consumer 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 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 record(String topic, int partition, String country, int population) { + return new ConsumerRecord<>(topic, partition, 0, country, population); + } +} \ No newline at end of file