Add unit tests

Signed-off-by: Chase Engelbrecht <[email protected]>
opensearch-project · Aug 17, 2023 · 53b4675 · 53b4675
1 parent e01f10b
commit 53b4675
Showing 1 changed file with 38 additions and 0 deletions.
diff --git a/...s/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/s3/S3SinkServiceTest.java b/...s/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/s3/S3SinkServiceTest.java
@@ -321,6 +321,44 @@ void test_output_with_uploadedToS3_failure_does_not_record_byte_count() throws I
         verify(buffer, times(3)).flushToS3(any(), anyString(), anyString());
     }
 
+    @Test
+    void test_output_with_no_incoming_records_flushes_batch() throws IOException {
+
+        bufferFactory = mock(BufferFactory.class);
+        Buffer buffer = mock(Buffer.class);
+        when(bufferFactory.getBuffer()).thenReturn(buffer);
+        when(buffer.getEventCount()).thenReturn(10);
+
+        final OutputStream outputStream = mock(OutputStream.class);
+        final Event event = JacksonEvent.fromMessage(UUID.randomUUID().toString());
+        doNothing().when(codec).writeEvent(event, outputStream);
+        final S3SinkService s3SinkService = createObjectUnderTest();
+        s3SinkService.output(Collections.emptyList());
+
+        verify(snapshotSuccessCounter, times(1)).increment();
+        verify(buffer, times(1)).flushToS3(any(), anyString(), anyString());
+    }
+
+    @Test
+    void test_output_with_no_incoming_records_or_buffered_records_short_circuits() throws IOException {
+
+        bufferFactory = mock(BufferFactory.class);
+        Buffer buffer = mock(Buffer.class);
+        when(bufferFactory.getBuffer()).thenReturn(buffer);
+        when(buffer.getEventCount()).thenReturn(0);
+        final long objectSize = random.nextInt(1_000_000) + 10_000;
+        when(buffer.getSize()).thenReturn(objectSize);
+
+        final OutputStream outputStream = mock(OutputStream.class);
+        final Event event = JacksonEvent.fromMessage(UUID.randomUUID().toString());
+        doNothing().when(codec).writeEvent(event, outputStream);
+        final S3SinkService s3SinkService = createObjectUnderTest();
+        s3SinkService.output(Collections.emptyList());
+
+        verify(snapshotSuccessCounter, times(0)).increment();
+        verify(buffer, times(0)).flushToS3(any(), anyString(), anyString());
+    }
+
     @Test
     void test_retryFlushToS3_positive() throws InterruptedException, IOException {