Add integration test for Ip2GeoProcessor

Signed-off-by: Heemin Kim <[email protected]>

src/main/java/org/opensearch/geospatial/exceptions/ResourceInUseException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.geospatial.exceptions;
+
+import java.io.IOException;
+
+import org.opensearch.OpenSearchException;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.rest.RestStatus;
+
+/**
+ * Generic ResourceInUseException corresponding to the {@link RestStatus#BAD_REQUEST} status code
+ */
+public class ResourceInUseException extends OpenSearchException {
+
+    public ResourceInUseException(String msg, Object... args) {
+        super(msg, args);
+    }
+
+    public ResourceInUseException(String msg, Throwable cause, Object... args) {
+        super(msg, cause, args);
+    }
+
+    public ResourceInUseException(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public final RestStatus status() {
+        return RestStatus.BAD_REQUEST;
+    }
+}

src/main/java/org/opensearch/geospatial/ip2geo/action/DeleteDatasourceTransportAction.java

@@ -17,6 +17,7 @@
 import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.geospatial.annotation.VisibleForTesting;
+import org.opensearch.geospatial.exceptions.ResourceInUseException;
 import org.opensearch.geospatial.ip2geo.common.DatasourceFacade;
 import org.opensearch.geospatial.ip2geo.common.DatasourceState;
 import org.opensearch.geospatial.ip2geo.common.Ip2GeoLockService;
@@ -105,7 +106,7 @@ protected void deleteDatasource(final String datasourceName) throws IOException
 
     private void setDatasourceStateAsDeleting(final Datasource datasource) {
         if (ip2GeoProcessorFacade.getProcessors(datasource.getName()).isEmpty() == false) {
-            throw new OpenSearchException("datasource is being used by one of processors");
+            throw new ResourceInUseException("datasource is being used by one of processors");
         }
 
         DatasourceState previousState = datasource.getState();
@@ -119,7 +120,7 @@ private void setDatasourceStateAsDeleting(final Datasource datasource) {
         if (ip2GeoProcessorFacade.getProcessors(datasource.getName()).isEmpty() == false) {
             datasource.setState(previousState);
             datasourceFacade.updateDatasource(datasource);
-            throw new OpenSearchException("datasource is being used by one of processors");
+            throw new ResourceInUseException("datasource is being used by one of processors");
         }
     }
 }

src/main/java/org/opensearch/geospatial/ip2geo/action/PutDatasourceRequest.java

@@ -36,9 +36,9 @@
 @Log4j2
 @EqualsAndHashCode
 public class PutDatasourceRequest extends AcknowledgedRequest<PutDatasourceRequest> {
-    private static final ParseField ENDPOINT_FIELD = new ParseField("endpoint");
-    private static final ParseField UPDATE_INTERVAL_IN_DAYS_FIELD = new ParseField("update_interval_in_days");
     private static final int MAX_DATASOURCE_NAME_BYTES = 255;
+    public static final ParseField ENDPOINT_FIELD = new ParseField("endpoint");
+    public static final ParseField UPDATE_INTERVAL_IN_DAYS_FIELD = new ParseField("update_interval_in_days");
     /**
      * @param name the datasource name
      * @return the datasource name

src/test/java/org/opensearch/geospatial/GeospatialRestTestCase.java

@@ -17,9 +17,12 @@
 import static org.opensearch.search.aggregations.Aggregations.AGGREGATIONS_FIELD;
 
 import java.io.IOException;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.IntStream;
@@ -46,12 +49,12 @@
 import org.opensearch.geospatial.action.upload.geojson.UploadGeoJSONRequestContent;
 import org.opensearch.geospatial.index.mapper.xyshape.XYShapeFieldMapper;
 import org.opensearch.geospatial.index.query.xyshape.XYShapeQueryBuilder;
+import org.opensearch.geospatial.ip2geo.common.DatasourceState;
 import org.opensearch.geospatial.rest.action.upload.geojson.RestUploadGeoJSONAction;
 import org.opensearch.ingest.Pipeline;
 import org.opensearch.rest.RestStatus;
 
 public abstract class GeospatialRestTestCase extends OpenSearchSecureRestTestCase {
-
     public static final String SOURCE = "_source";
     public static final String DOC = "_doc";
     public static final String URL_DELIMITER = "/";
@@ -69,11 +72,23 @@ public abstract class GeospatialRestTestCase extends OpenSearchSecureRestTestCas
     public static final String SHAPE_ID_FIELD = "id";
     public static final String SHAPE_INDEX_PATH_FIELD = "path";
     public static final String QUERY_PARAM_TOKEN = "?";
+    private static final String SETTINGS = "_settings";
+    private static final String SIMULATE = "_simulate";
+    private static final String DOCS = "docs";
+    private static final String DATASOURCES = "datasources";
+    private static final String STATE = "state";
+    private static final String PUT = "PUT";
+    private static final String GET = "GET";
+    private static final String DELETE = "DELETE";
 
     private static String buildPipelinePath(String name) {
         return String.join(URL_DELIMITER, "_ingest", "pipeline", name);
     }
 
+    private static String buildDatasourcePath(String name) {
+        return String.join(URL_DELIMITER, getPluginURLPrefix(), "ip2geo/datasource", name);
+    }
+
     protected static Response createPipeline(String name, Optional<String> description, List<Map<String, Object>> processorConfigs)
         throws IOException {
         XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
@@ -95,6 +110,69 @@ protected static void deletePipeline(String name) throws IOException {
         client().performRequest(request);
     }
 
+    protected Response createDatasource(final String name, Map<String, String> properties) throws IOException {
+        XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
+        for (Map.Entry<String, String> config : properties.entrySet()) {
+            builder.field(config.getKey(), config.getValue());
+        }
+        builder.endObject();
+
+        Request request = new Request(PUT, buildDatasourcePath(name));
+        request.setJsonEntity(Strings.toString(builder));
+        return client().performRequest(request);
+    }
+
+    protected void waitForDatasourceStateTobe(final String name, final DatasourceState state, final Duration timeout) throws Exception {
+        Instant start = Instant.now();
+        while (state.equals(getDatasourceState(name)) == false) {
+            if (Duration.between(start, Instant.now()).compareTo(timeout) > 0) {
+                throw new RuntimeException(
+                    String.format(Locale.ROOT, "Datasource state didn't change to %s after %d seconds", state.name(), timeout.toSeconds())
+                );
+            }
+            Thread.sleep(1000);
+        }
+    }
+
+    private DatasourceState getDatasourceState(final String name) throws Exception {
+        List<Map<String, Object>> datasources = (List<Map<String, Object>>) getDatasource(name).get(DATASOURCES);
+        return DatasourceState.valueOf((String) datasources.get(0).get(STATE));
+    }
+
+    protected Response deleteDatasource(final String name) throws IOException {
+        Request request = new Request(DELETE, buildDatasourcePath(name));
+        return client().performRequest(request);
+    }
+
+    protected Map<String, Object> getDatasource(final String name) throws Exception {
+        Request request = new Request(GET, buildDatasourcePath(name));
+        Response response = client().performRequest(request);
+        return createParser(XContentType.JSON.xContent(), EntityUtils.toString(response.getEntity())).map();
+    }
+
+    protected Response updateDatasource(final String name, Map<String, Object> config) throws IOException {
+        XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
+        if (config != null && !config.isEmpty()) {
+            builder.value(config);
+        }
+        builder.endObject();
+
+        Request request = new Request(PUT, String.join(URL_DELIMITER, buildDatasourcePath(name), SETTINGS));
+        request.setJsonEntity(Strings.toString(builder));
+        return client().performRequest(request);
+    }
+
+    protected Map<String, Object> simulatePipeline(final String name, List<Object> docs) throws Exception {
+        XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
+        builder.field(DOCS, docs);
+        builder.endObject();
+
+        Request request = new Request(GET, String.join(URL_DELIMITER, buildPipelinePath(name), SIMULATE));
+        request.setJsonEntity(Strings.toString(builder));
+        Response response = client().performRequest(request);
+        return createParser(XContentType.JSON.xContent(), EntityUtils.toString(response.getEntity())).map();
+    }
+
     protected static void createIndex(String name, Settings settings, Map<String, String> fieldMap) throws IOException {
         XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().startObject().startObject(MAPPING_PROPERTIES_KEY);
         for (Map.Entry<String, String> entry : fieldMap.entrySet()) {
@@ -136,7 +214,7 @@ public static String indexDocument(String indexName, String docID, String body,
         return docID;
     }
 
-    protected Map<String, Object> buildProcessorConfig(final String processorType, final Map<String, String> properties) {
+    protected Map<String, Object> buildProcessorConfig(final String processorType, final Map<String, Object> properties) {
         Map<String, Object> featureProcessor = new HashMap<>();
         featureProcessor.put(processorType, properties);
         return featureProcessor;

src/test/java/org/opensearch/geospatial/ip2geo/Ip2GeoTestCase.java

@@ -179,11 +179,11 @@ protected long randomPositiveLong() {
     }
 
     protected Datasource randomDatasource() {
-        int validForInDays = Randomness.get().nextInt(30) + 2;
+        int validForInDays = Randomness.get().nextInt(30) + 3;
         Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS);
         Datasource datasource = new Datasource();
         datasource.setName(GeospatialTestHelper.randomLowerCaseString());
-        datasource.setSchedule(new IntervalSchedule(now, validForInDays - 1, ChronoUnit.DAYS));
+        datasource.setSchedule(new IntervalSchedule(now, validForInDays - 2, ChronoUnit.DAYS));
         datasource.setState(randomState());
         datasource.setIndices(Arrays.asList(GeospatialTestHelper.randomLowerCaseString(), GeospatialTestHelper.randomLowerCaseString()));
         datasource.setEndpoint(String.format(Locale.ROOT, "https://%s.com/manifest.json", GeospatialTestHelper.randomLowerCaseString()));