Fix #3459 Unit tests fail on Windows machine (#3461)

1. SinkModelTest: Use system System.lineSeparator() instead of hardcode '\n' 2. DataPrepperArgsTest: Covert file path separators to local system. 3. DateProcessorTests: Covert time to same timezone before comparing. 4. InMemorySourceCoordinationStoreTest: Use greaterThanOrEqualTo to compare time since they may be same. 5. QueuedPartitionsItemTest: Use sleep to get two different time instances. 6. RSSSourceTest: Use mocker server to avoid internet connecting. 7. ParquetOutputCodecTest: Close all outputStream objects in the tests. 8. org.opensearch.dataprepper.plugins.sink.s3.accumulator.InMemoryBufferTest#getDuration_provides_duration_within_expected_range: No solution to fix. Disable it. Please see my comments in the test file. Signed-off-by: Gong Yi <[email protected]>
opensearch-project · Oct 10, 2023 · 6e380d0 · 6e380d0
1 parent d3179f0
commit 6e380d0
Show file tree

Hide file tree

Showing 11 changed files with 1,486 additions and 18 deletions.
diff --git a/...epper-api/src/test/java/org/opensearch/dataprepper/model/configuration/SinkModelTest.java b/...epper-api/src/test/java/org/opensearch/dataprepper/model/configuration/SinkModelTest.java
@@ -87,7 +87,7 @@ void serialize_into_known_SinkModel() throws IOException {
 
         final String expectedJson = createStringFromInputStream(this.getClass().getResourceAsStream("sink_plugin.yaml"));
 
-        assertThat("---\n" + actualJson, equalTo(expectedJson));
+        assertThat("---" + System.lineSeparator() + actualJson, equalTo(expectedJson));
         assertThat(sinkModel.getTagsTargetKey(), equalTo(tagsTargetKey));
 
     }
@@ -140,7 +140,7 @@ void serialize_with_just_pluginModel() throws IOException {
 
         final String expectedJson = createStringFromInputStream(this.getClass().getResourceAsStream("/serialized_with_plugin_settings.yaml"));
 
-        assertThat("---\n" + actualJson, equalTo(expectedJson));
+        assertThat("---" + System.lineSeparator() + actualJson, equalTo(expectedJson));
     }
 
     @Test

diff --git a/data-prepper-main/build.gradle b/data-prepper-main/build.gradle
@@ -18,6 +18,7 @@ dependencies {
     implementation(libs.spring.context) {
         exclude group: 'commons-logging', module: 'commons-logging'
     }
+    testImplementation 'commons-io:commons-io:2.14.0'
 }
 
 jar {

diff --git a/data-prepper-main/src/test/java/org/opensearch/dataprepper/DataPrepperArgsTest.java b/data-prepper-main/src/test/java/org/opensearch/dataprepper/DataPrepperArgsTest.java
@@ -11,6 +11,7 @@
 
 import java.nio.file.Paths;
 
+import static org.apache.commons.io.FilenameUtils.separatorsToSystem;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.is;
@@ -84,7 +85,7 @@ void testGivenLogstashConfigDirectoryThenPipelineConfigCreated() {
         final String configFile = Paths.get("src", "test", "resources", "logstash-conf/").toString();
 
         assertThat(args, is(notNullValue()));
-        assertThat(args.getPipelineConfigFileLocation(), is(configFile));
+        assertThat(separatorsToSystem(args.getPipelineConfigFileLocation()), is(separatorsToSystem(configFile)));
         assertThat(args.getDataPrepperConfigFileLocation(), is(DP_CONFIG_YAML_FILE_PATH));
     }
 

diff --git a/...r/src/test/java/org/opensearch/dataprepper/plugins/processor/date/DateProcessorTests.java b/...r/src/test/java/org/opensearch/dataprepper/plugins/processor/date/DateProcessorTests.java
@@ -353,7 +353,8 @@ void match_with_different_year_formats_test(String pattern) {
         final Record<Event> record = buildRecordWithEvent(testData);
         final List<Record<Event>> processedRecords = (List<Record<Event>>) dateProcessor.doExecute(Collections.singletonList(record));
 
-        ZonedDateTime actualZonedDateTime =  record.getData().get(TIMESTAMP_KEY, ZonedDateTime.class);
+        //The timezone from 'record' instance is UTC instead of local one. We need convert it to local.
+        ZonedDateTime actualZonedDateTime =  record.getData().get(TIMESTAMP_KEY, ZonedDateTime.class).withZoneSameInstant(mockDateProcessorConfig.getSourceZoneId());
         ZonedDateTime expectedZonedDatetime = expectedDateTime.minus(10, ChronoUnit.YEARS).atZone(mockDateProcessorConfig.getSourceZoneId()).truncatedTo(ChronoUnit.SECONDS);
 
         Assertions.assertTrue(actualZonedDateTime.toLocalDate().isEqual(expectedZonedDatetime.toLocalDate()));
@@ -380,7 +381,8 @@ void match_without_year_test(String pattern) {
         final Record<Event> record = buildRecordWithEvent(testData);
         final List<Record<Event>> processedRecords = (List<Record<Event>>) dateProcessor.doExecute(Collections.singletonList(record));
 
-        ZonedDateTime actualZonedDateTime =  record.getData().get(TIMESTAMP_KEY, ZonedDateTime.class);
+        //The timezone from record is UTC instead of local one. We need convert it to local.
+        ZonedDateTime actualZonedDateTime =  record.getData().get(TIMESTAMP_KEY, ZonedDateTime.class).withZoneSameInstant(mockDateProcessorConfig.getSourceZoneId());
         ZonedDateTime expectedZonedDatetime = expectedDateTime.atZone(mockDateProcessorConfig.getSourceZoneId()).truncatedTo(ChronoUnit.SECONDS);
 
         Assertions.assertTrue(actualZonedDateTime.toLocalDate().isEqual(expectedZonedDatetime.toLocalDate()));

diff --git a/...h/dataprepper/plugins/sourcecoordinator/inmemory/InMemorySourceCoordinationStoreTest.java b/...h/dataprepper/plugins/sourcecoordinator/inmemory/InMemorySourceCoordinationStoreTest.java
@@ -20,7 +20,7 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.doNothing;
@@ -68,7 +68,7 @@ void tryAcquireAvailablePartition_returns_empty_optional_when_no_item_is_availab
     }
 
     @Test
-    void tryAcquireAvailablePartition_gets_item_from_inMemoryPartitionAccessor_and_modifies_it_correctly() {
+    void tryAcquireAvailablePartition_gets_item_from_inMemoryPartitionAccessor_and_modifies_it_correctly() throws InterruptedException {
         final String sourceIdentifier = UUID.randomUUID().toString();
         final String ownerId = UUID.randomUUID().toString();
         final Duration ownershipTimeout = Duration.ofMinutes(2);
@@ -79,12 +79,11 @@ void tryAcquireAvailablePartition_gets_item_from_inMemoryPartitionAccessor_and_m
         given(inMemoryPartitionAccessor.getNextItem()).willReturn(Optional.of(item));
 
         final InMemorySourceCoordinationStore objectUnderTest = createObjectUnderTest();
-
         final Optional<SourcePartitionStoreItem> result = objectUnderTest.tryAcquireAvailablePartition(sourceIdentifier, ownerId, ownershipTimeout);
         assertThat(result.isPresent(), equalTo(true));
         assertThat(result.get(), equalTo(item));
         assertThat(result.get().getSourcePartitionStatus(), equalTo(SourcePartitionStatus.ASSIGNED));
-        assertThat(result.get().getPartitionOwnershipTimeout(), greaterThan(now.plus(ownershipTimeout)));
+        assertThat(result.get().getPartitionOwnershipTimeout(), greaterThanOrEqualTo(now.plus(ownershipTimeout)));
         assertThat(result.get().getPartitionOwner(), equalTo(ownerId));
 
     }

diff --git a/...g/opensearch/dataprepper/plugins/sourcecoordinator/inmemory/QueuedPartitionsItemTest.java b/...g/opensearch/dataprepper/plugins/sourcecoordinator/inmemory/QueuedPartitionsItemTest.java
@@ -54,21 +54,20 @@ void queued_partitions_items_with_one_null_sortedTimestamp_and_one_non_null_cons
     }
 
     @Test
-    void queued_partitions_item_compares_based_on_sorted_timestamp() {
+    void queued_partitions_item_compares_based_on_sorted_timestamp() throws InterruptedException {
         final InMemoryPartitionAccessor.QueuedPartitionsItem firstItem = new InMemoryPartitionAccessor.QueuedPartitionsItem(
                 UUID.randomUUID().toString(),
                 UUID.randomUUID().toString(),
                 Instant.now()
         );
-
+        Thread.sleep(100);
         final Instant now = Instant.now();
 
         final InMemoryPartitionAccessor.QueuedPartitionsItem secondItem = new InMemoryPartitionAccessor.QueuedPartitionsItem(
                 UUID.randomUUID().toString(),
                 UUID.randomUUID().toString(),
                 now
         );
-
         final InMemoryPartitionAccessor.QueuedPartitionsItem thirdItem = new InMemoryPartitionAccessor.QueuedPartitionsItem(
                 UUID.randomUUID().toString(),
                 UUID.randomUUID().toString(),

diff --git a/data-prepper-plugins/rss-source/build.gradle b/data-prepper-plugins/rss-source/build.gradle
@@ -20,6 +20,8 @@ dependencies {
     implementation 'com.apptasticsoftware:rssreader:3.2.5'
     testImplementation libs.commons.lang3
     testImplementation project(':data-prepper-test-common')
+    testImplementation 'org.mock-server:mockserver-junit-jupiter-no-dependencies:5.14.0'
+    testImplementation 'commons-io:commons-io:2.14.0'
 }
 
 test {

diff --git a/...rss-source/src/test/java/org/opensearch/dataprepper/plugins/source/rss/RSSSourceTest.java b/...rss-source/src/test/java/org/opensearch/dataprepper/plugins/source/rss/RSSSourceTest.java
@@ -5,18 +5,24 @@
 
 package org.opensearch.dataprepper.plugins.source.rss;
 
+import org.apache.commons.io.IOUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.junit.jupiter.MockServerExtension;
+import org.mockserver.model.MediaType;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.buffer.Buffer;
 import org.opensearch.dataprepper.model.document.Document;
 import org.opensearch.dataprepper.model.record.Record;
 
+import java.io.IOException;
 import java.time.Duration;
+import java.util.Map;
 
 import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.anyCollection;
@@ -25,16 +31,17 @@
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.verify;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
 
 @ExtendWith(MockitoExtension.class)
+@ExtendWith(MockServerExtension.class)
 class RSSSourceTest {
 
     private final String PLUGIN_NAME = "rss";
 
     private final String PIPELINE_NAME = "test";
 
-    private final String VALID_RSS_URL = "https://forum.opensearch.org/latest.rss";
-
     @Mock
     private Buffer<Record<Document>> buffer;
 
@@ -46,13 +53,37 @@ class RSSSourceTest {
     private RSSSource rssSource;
     private Duration pollingFrequency;
 
+    private final ClientAndServer client;
+    private final String rssMockUrl;
+
+    public RSSSourceTest(ClientAndServer client) {
+        this.client = client;
+        this.rssMockUrl= "http://localhost:"+client.getPort()+"/latest.rss";
+
+    }
+
+
     @BeforeEach
-    void setUp() {
+    void setUp() throws IOException {
         pluginMetrics = PluginMetrics.fromNames(PLUGIN_NAME, PIPELINE_NAME);
         pollingFrequency = Duration.ofMillis(1800);
-        lenient().when(rssSourceConfig.getUrl()).thenReturn(VALID_RSS_URL);
+        lenient().when(rssSourceConfig.getUrl()).thenReturn(rssMockUrl);
         lenient().when(rssSourceConfig.getPollingFrequency()).thenReturn(pollingFrequency);
         rssSource = new RSSSource(pluginMetrics, rssSourceConfig);
+        client.reset();
+        var rssContent = IOUtils.resourceToByteArray("rss.xml",RSSSourceTest.class.getClassLoader());
+        client
+                .when(
+                        request()
+                                .withMethod("GET")
+                                .withPath("/latest.rss")
+                )
+                .respond(
+                        response()
+                                .withContentType(new MediaType("application", "rss+xml", Map.of("charset","utf-8")))
+                                .withBody(rssContent)
+
+                );
     }
 
     @AfterEach