[SOY] Operation non it

services/api-event-donneesentrees/src/test/java/org/mte/numecoeval/donneesentrees/IntegrationOperationNonITTest.java

+package org.mte.numecoeval.donneesentrees;
+
+import io.zonky.test.db.AutoConfigureEmbeddedDatabase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mte.numecoeval.donneesentrees.test.jdbc.ScriptUtils;
+import org.mte.numecoeval.donneesentrees.test.kafka.KafkaConsumerOperationNonIT;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.await;
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@ActiveProfiles(profiles = {"test"})
+@AutoConfigureEmbeddedDatabase
+@EmbeddedKafka(
+        bootstrapServersProperty = "spring.kafka.bootstrap-servers",
+        partitions = 1
+)
+class IntegrationOperationNonITTest {
+
+    @Autowired
+    JdbcTemplate jdbcTemplate;
+
+    @Autowired
+    KafkaConsumerOperationNonIT kafkaConsumer;
+
+    @BeforeEach
+    void setup() {
+        ScriptUtils.loadScript(jdbcTemplate.getDataSource(), "sql/schema.sql");
+        jdbcTemplate.batchUpdate("DELETE FROM en_operation_non_it");
+    }
+
+    @Test
+    void whenDataAvailable_shouldUpdateStatusAndSendMessage() {
+        ScriptUtils.loadScript(jdbcTemplate.getDataSource(), "sql/operation_non_it.sql");
+
+        // Given
+        var queryResultCountEntrees = jdbcTemplate.query(
+                "SELECT count(*) as nbr from en_operation_non_it",
+                (rs, rowNum) -> rs.getInt("nbr")
+        );
+        var nbrEntrees = queryResultCountEntrees.get(0);
+
+        // When
+        jdbcTemplate.batchUpdate("UPDATE en_operation_non_it SET statut_traitement = 'A_INGERER'");
+
+        // Then
+        Assertions.assertEquals(1, nbrEntrees);
+        await().atMost(10, TimeUnit.SECONDS)
+                .until(() -> {
+                    var queryResultCountEntreesIngerees = jdbcTemplate.query(
+                            "SELECT count(*) as nbr from en_operation_non_it WHERE statut_traitement = 'INGERE'",
+                            (rs, rowNum) -> rs.getInt("nbr")
+                    );
+
+                    return !queryResultCountEntreesIngerees.isEmpty() && Objects.equals(nbrEntrees, queryResultCountEntreesIngerees.get(0));
+                });
+
+        await().atMost(5, TimeUnit.SECONDS)
+                .until(() -> nbrEntrees == kafkaConsumer.getPayloads().size());
+
+    }
+
+}