Trading bot db tests

trading-bot/src/main/java/ru/ct/belfort/db/IdeasRepository.java

@@ -24,25 +24,24 @@ public IdeasRepository(DataSource dataSource) {
 
     public void insert(IdeaDTO idea) {
         jdbcTemplate.update("""
-            INSERT INTO ideas(score, time) VALUES(?, CURRENT_TIMESTAMP(0))
+            INSERT INTO ideas(score, time) VALUES(?, CURRENT_TIMESTAMP)
         """, idea.score());
-        debugOutput();
     }
 
     public List<IdeaEntity> selectAll() {
         return jdbcTemplate.query("SELECT * FROM ideas", mapper);
     }
 
-    public void deleteAll() {
-        jdbcTemplate.update("DELETE FROM ideas");
+    public Integer getRecordsAmount() {
+        return jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ideas", Integer.class);
     }
 
-    // TODO: Remove this when db tests appear
-    private void debugOutput() {
-        log.info("Inserted");
-        for (IdeaEntity it : selectAll()) {
-            log.info(it.toString());
-        }
+    public IdeaEntity getLastRecord() {
+        return jdbcTemplate.query("SELECT * FROM ideas ORDER BY time DESC LIMIT 1", mapper).get(0);
+    }
+
+    public void deleteAll() {
+        jdbcTemplate.update("DELETE FROM ideas");
     }
 
     private static class IdeaEntityMapper implements RowMapper<IdeaEntity> {

trading-bot/src/test/java/ru/ct/belfort/Utils.java

@@ -39,4 +39,32 @@ public static TradingInfoDTO genRandomTradingInfoDTO(String strategy) {
         var candles = genRandomCandles(10, 5, 10);
         return new TradingInfoDTO(candles, strategy);
     }
+
+    public static TradingInfoDTO badCandlesSample() {
+        return new TradingInfoDTO(
+                closePricesToCandles(new double[]{500, 493, 491, 485, 483, 482, 480, 467, 463, 461}),
+                "rsi"
+        );
+    }
+
+    public static TradingInfoDTO goodCandlesSample() {
+        return new TradingInfoDTO(
+                closePricesToCandles(new double[]{500, 505, 517, 523, 524, 525, 536, 541, 547, 555}),
+                "rsi"
+        );
+    }
+
+    public static TradingInfoDTO okCandlesSample() {
+        return new TradingInfoDTO(
+                closePricesToCandles(new double[]{500, 500, 500, 500, 500, 500, 500, 500, 500, 500}),
+                "rsi"
+        );
+    }
+
+    public static TradingInfoDTO unknownStrategySample() {
+        return new TradingInfoDTO(
+                closePricesToCandles(new double[]{}),
+                "fds7f757das78f7sd"
+        );
+    }
 }

trading-bot/src/test/java/ru/ct/belfort/base/KafkaAndPostgresTestBase.java

@@ -0,0 +1,93 @@
+package ru.ct.belfort.base;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.BeforeAll;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.PostgreSQLContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import ru.ct.belfort.IdeaDTO;
+import ru.ct.belfort.TradingInfoDTO;
+import ru.ct.belfort.kafka.producers.ErrorProducerConfig;
+import ru.ct.belfort.kafka.producers.IdeasProducerConfig;
+
+import java.util.Collections;
+import java.util.Map;
+
+@SpringBootTest
+@Testcontainers
+@DirtiesContext
+public abstract class KafkaAndPostgresTestBase {
+
+    @Container
+    protected static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"));
+
+    @Container
+    protected static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres");
+
+    @DynamicPropertySource
+    static void kafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
+        registry.add("spring.datasource.drivername", postgres::getDriverClassName);
+        registry.add("spring.datasource.url", postgres::getJdbcUrl);
+        registry.add("spring.datasource.username", postgres::getUsername);
+        registry.add("spring.datasource.password", postgres::getPassword);
+    }
+
+    protected static KafkaProducer<String, TradingInfoDTO> producer = null;
+    protected static KafkaConsumer<String, IdeaDTO> ideasConsumer = null;
+    protected static KafkaConsumer<String, String> errorConsumer = null;
+
+    @BeforeAll
+    public static void init() {
+
+        JsonDeserializer<IdeaDTO> jsonDeserializer = new JsonDeserializer<>();
+        jsonDeserializer.addTrustedPackages("*");
+
+        producer = new KafkaProducer<>(
+                Map.of(
+                        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers(),
+                        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class,
+                        ProducerConfig.CLIENT_ID_CONFIG, "candles_test_producer"
+                ),
+                new StringSerializer(),
+                new JsonSerializer<>()
+        );
+
+        ideasConsumer = new KafkaConsumer<>(
+                Map.of(
+                        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers(),
+                        ConsumerConfig.GROUP_ID_CONFIG, "ideas_test_consumers",
+                        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
+                ),
+                new StringDeserializer(),
+                jsonDeserializer
+        );
+
+        errorConsumer = new KafkaConsumer<>(
+                Map.of(
+                        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers(),
+                        ConsumerConfig.GROUP_ID_CONFIG, "error_test_consumers",
+                        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
+                ),
+                new StringDeserializer(),
+                new StringDeserializer()
+        );
+
+        ideasConsumer.subscribe(Collections.singletonList(IdeasProducerConfig.TOPIC));
+        errorConsumer.subscribe(Collections.singletonList(ErrorProducerConfig.TOPIC));
+    }
+}

trading-bot/src/test/java/ru/ct/belfort/db/PostgresTest.java

@@ -0,0 +1,65 @@
+package ru.ct.belfort.db;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import ru.ct.belfort.TradingInfoDTO;
+import ru.ct.belfort.base.KafkaAndPostgresTestBase;
+import ru.ct.belfort.kafka.consumers.CandlesConsumerConfig;
+import ru.ct.belfort.strategy.RsiStrategy;
+
+import java.time.Duration;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static ru.ct.belfort.Utils.*;
+
+@Testcontainers
+@SpringBootTest
+public class PostgresTest extends KafkaAndPostgresTestBase {
+
+    @Autowired
+    IdeasRepository repository;
+
+    private static final long MAX_BLOCK_TIME = 1000;
+
+    private static void sendAndWait(TradingInfoDTO message) {
+        producer.send(new ProducerRecord<>(CandlesConsumerConfig.TOPIC, message));
+        ideasConsumer.poll(Duration.ofMillis(MAX_BLOCK_TIME));
+    }
+
+    @Test
+    public void sendBadCandles_ExpectDBInsertion() {
+        final var message = badCandlesSample();
+        var before = repository.getRecordsAmount();
+        sendAndWait(message);
+        assertEquals(before + 1, repository.getRecordsAmount());
+        double score = new RsiStrategy().predict(message.candles());
+        assertEquals(repository.getLastRecord().score(), score);
+    }
+
+    @Test
+    public void sendGoodCandles_ExpectDBInsertion() {
+        final var message = goodCandlesSample();
+        var before = repository.getRecordsAmount();
+        sendAndWait(message);
+        assertEquals(before + 1, repository.getRecordsAmount());
+        double score = new RsiStrategy().predict(message.candles());
+        assertEquals(repository.getLastRecord().score(), score);
+    }
+
+    @Test
+    public void sendOkCandles_ExpectNoDBInsertion() {
+        var before = repository.getRecordsAmount();
+        sendAndWait(okCandlesSample());
+        assertEquals(before, repository.getRecordsAmount());
+    }
+
+    @Test
+    public void sendUnknownStrategy_ExpectNoDBInsertion() {
+        var before = repository.getRecordsAmount();
+        sendAndWait(unknownStrategySample());
+        assertEquals(before, repository.getRecordsAmount());
+    }
+}