From 23b471f93914ecc866582bc83fd43b981aa4912a Mon Sep 17 00:00:00 2001
From: Paul Ebermann <paul.ebermann@zalando.de>
Date: Fri, 28 Jul 2023 19:48:37 +0200
Subject: [PATCH] add test for deletion

---
 .../eventlog/impl/EventLogRepositoryIT.java   | 37 ++++++++++++++++++-
 1 file changed, 36 insertions(+), 1 deletion(-)

diff --git a/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryIT.java b/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryIT.java
index c8e887f..4370e1e 100644
--- a/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryIT.java
+++ b/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryIT.java
@@ -1,6 +1,7 @@
 package org.zalando.nakadiproducer.eventlog.impl;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.core.Is.is;
 
 import javax.transaction.Transactional;
@@ -8,9 +9,12 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.BeanPropertyRowMapper;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.zalando.nakadiproducer.BaseMockedExternalCommunicationIT;
 
+import java.util.List;
+
 @Transactional
 public class EventLogRepositoryIT extends BaseMockedExternalCommunicationIT {
 
@@ -44,14 +48,45 @@ public class EventLogRepositoryIT extends BaseMockedExternalCommunicationIT {
     public void setUp() {
         eventLogRepository.deleteAll();
 
+        persistTestEvent("FLOW_ID");
+    }
+
+    private void persistTestEvent(String flowId) {
         final EventLog eventLog = EventLog.builder()
                 .eventBodyData(WAREHOUSE_EVENT_BODY_DATA)
                 .eventType(WAREHOUSE_EVENT_TYPE)
                 .compactionKey(COMPACTION_KEY)
-                .flowId("FLOW_ID").build();
+                .flowId(flowId)
+                .build();
         eventLogRepository.persist(eventLog);
     }
 
+    @Test
+    public void testDeleteMultipleEvents() {
+        persistTestEvent("second_Flow-ID");
+        persistTestEvent("third flow-ID");
+        persistTestEvent("fourth flow-ID");
+        persistTestEvent("fifth flow-ID");
+
+        List<EventLog> events = findAllEventsInDB();
+        assertThat(events, hasSize(5));
+        EventLog notDeleted = events.remove(0);
+
+        // now the actual test – delete just 4 of the 5 events from the DB
+        eventLogRepository.delete(events);
+
+        List<EventLog> remaining = findAllEventsInDB();
+        assertThat(remaining, hasSize(1));
+        assertThat(remaining.get(0).getId(), is(notDeleted.getId()));
+        assertThat(remaining.get(0).getFlowId(), is(notDeleted.getFlowId()));
+    }
+
+    private List<EventLog> findAllEventsInDB() {
+        return jdbcTemplate.query(
+                "SELECT * FROM nakadi_events.event_log",
+                new BeanPropertyRowMapper<>(EventLog.class));
+    }
+
     @Test
     public void testFindEventInRepositoryById() {
         Integer id = jdbcTemplate.queryForObject(