Aws add schema tests

NOTICE

@@ -499,13 +499,6 @@ The following software have components provided under the terms of this license:
 - Jakarta XML Binding API (from https://repo1.maven.org/maven2/jakarta/xml/bind/jakarta.xml.bind-api, https://repo1.maven.org/maven2/org/jboss/spec/javax/xml/bind/jboss-jaxb-api_2.3_spec)
 - Old JAXB Core (from <https://eclipse-ee4j.github.io/jaxb-ri/>, https://eclipse-ee4j.github.io/jaxb-ri/, https://repo1.maven.org/maven2/com/sun/xml/bind/jaxb-impl)
 
-========================================================================
-GPL-2.0-only
-========================================================================
-The following software have components provided under the terms of this license:
-
-- javax.annotation API (from http://jcp.org/en/jsr/detail?id=250)
-
 ========================================================================
 GPL-2.0-with-classpath-exception
 ========================================================================

testing/notification-test-aws/build-aws/push-endpoint/setup_teardown_endpoint.py

@@ -433,6 +433,11 @@ class CommonParameters:
         sp.check_output(["kubectl", "apply", "-f", "-"], input=new_auth_policy,
                         encoding='utf-8')
 
+    def restart_istio_ingress_deployment(self):
+        sp.check_output(["kubectl", "rollout", "restart", "-n",
+                         self.ingress_namespace,
+                         f"deployment/istio-ingress-{self.osdu_instance}"])
+
     async def create_resources(self):
         received_queues = asyncio.Queue()
         tasks = []
@@ -464,6 +469,9 @@ class CommonParameters:
         print("Now changing the JWT Authorization Policy")
         self.disable_jwt_auth_on_path()
 
+        print("Now restarting the Istio Ingress Deployment")
+        self.restart_istio_ingress_deployment()
+
     async def ensure_queue_deleted(self, queue_num: int):
         queue_name = self.get_sqs_queue_name(queue_num)
         try:

testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/Application.java

@@ -15,11 +15,8 @@
 
 package org.example;
 
-import jakarta.servlet.http.HttpServlet;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.web.servlet.ServletRegistrationBean;
-import org.springframework.context.annotation.Bean;
 
 @SpringBootApplication
 public class Application {

testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/RecordUtils.java

@@ -31,31 +31,18 @@ public class RecordUtils {
 
     private static final String domain = System.getProperty("DOMAIN", System.getenv("DOMAIN"));
 
-    public static Record[] createRecordsWithReference(int recordsCount, String id, String kind, String legalTag, String fromCrs, String conversionType, boolean isTestPartition) {
+    public static Record[] createRecordsWithReference(int recordsCount, String id, String kind, String legalTag, boolean isTestPartition) {
 
         Record[] records = new Record[recordsCount];
 
         for (int i = 0; i < recordsCount; i++) {
 
             Map<String, Object> data = new HashMap<>();
-            data.put("X", 16.00);
-            data.put("Y", 10.00);
-            data.put("Z", 0.0);
+            String specificId = id + i;
+            data.put("name", specificId);
+            Map<String, Object>[] metaBlocks = new Map[0];
 
-            JsonArray propertyNames = new JsonArray();
-            propertyNames.add("X");
-            propertyNames.add("Y");
-            propertyNames.add("Z");
-
-            Map<String, Object> meta = new HashMap<>();
-            meta.put(Constants.KIND, conversionType);
-            meta.put(Constants.PERSISTABLE_REFERENCE, fromCrs);
-            meta.put(Constants.PROPERTY_NAMES, propertyNames);
-
-            Map<String, Object>[] metaBlocks = new Map[1];
-            metaBlocks[0] = meta;
-
-            Record record = getRecordWithInputData(id + i, kind, legalTag, data, isTestPartition);
+            Record record = getRecordWithInputData(specificId, kind, legalTag, data, isTestPartition);
             record.setMeta(metaBlocks);
 
             records[i] = record;

testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/SchemaTestUtils.java

+/* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.subscriptions;
+
+import static org.junit.Assert.assertEquals;
+
+import com.amazonaws.services.sqs.model.Message;
+import com.google.gson.Gson;
+import com.sun.jersey.api.client.Client;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.commons.lang3.tuple.Pair;
+import org.opengroup.osdu.core.common.http.DpsHeaderFactory;
+import org.opengroup.osdu.core.common.logging.DefaultLogger;
+import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.notification.util.AwsTestUtils;
+import org.opengroup.osdu.notification.util.TestUtils;
+
+import javax.ws.rs.core.MediaType;
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class SchemaTestUtils {
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class SchemaPubSubMessage {
+        private String kind;
+        private String op;
+    }
+    private final Client restClient;
+    private final DpsHeaders dpsHeaders;
+    private final String schemaApiBaseUrl;
+    private final JaxRsDpsLog logger;
+    public SchemaTestUtils(DpsHeaders headers) {
+        restClient = Client.create();
+        dpsHeaders = DpsHeaderFactory.createFromMap(headers.getHeaders());
+        dpsHeaders.addCorrelationIdIfMissing();
+        logger = new JaxRsDpsLog(new DefaultLogger(), dpsHeaders);
+        schemaApiBaseUrl = String.format("%s/api/schema-service/v1", System.getProperty("SCHEMA_URL", System.getenv("SCHEMA_URL")));
+    }
+
+    private static final String SCHEMA_ID = "%s:%s:%s:%d.%d.%d";
+    private static final String SCHEMA_BODY = """
+{
+    "schemaInfo":{
+      "schemaIdentity":{
+         "authority":"%s",
+         "source":"%s",
+         "entityType":"%s",
+         "schemaVersionMajor":%d,
+         "schemaVersionMinor":%d,
+         "schemaVersionPatch":%d,
+         "id":"%s"
+      },
+      "status":"DEVELOPMENT"
+    },
+    "schema":{
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "type": "object",
+        "title": "My schema",
+        "definitions": {},
+        "properties": {
+            "data": {
+                "allOf": [
+                    {
+                        "type": "object",
+                        "properties": {
+                            "name": { "type": "string" }
+                        }
+                    }
+                ]
+            }
+        }
+    }
+}
+""";
+    private Pair<String, String> generateSchemaIdAndBody(String tenant, String source, String entityType, int major, int minor, int patch) {
+        String id = String.format(SCHEMA_ID, tenant, source, entityType, major, minor, patch);
+        String body = String.format(SCHEMA_BODY, tenant, source, entityType, major, minor, patch, id);
+        return Pair.of(id, body);
+    }
+
+    public SchemaPubSubMessage createSchemaForTest() {
+        return new SchemaPubSubMessage(createSchema(), "create");
+    }
+
+    public String createSchema() {
+        long millis = Instant.now().toEpochMilli();
+        String entityType = String.format("test-entity-%d", millis);
+        Pair<String, String> idAndBody = generateSchemaIdAndBody(TestUtils.getOsduTenant(), "test", entityType, 0, 0, 0);
+        ClientResponse response = sendRequest(String.format("%s/schema", schemaApiBaseUrl), "POST", true, idAndBody.getRight(), null);
+
+        InputStream inputStream = response.getEntityInputStream();
+        String body = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)).lines().collect(Collectors.joining("\n"));
+        logger.info("Schema service returned the following");
+        logger.info(String.format("Full schema body: %s", idAndBody.getRight()));
+        logger.info(body);
+        assertEquals(201, response.getStatus());
+        return idAndBody.getLeft();
+    }
+
+    public void validate(SchemaPubSubMessage expected, List<Message> receivedMessages) throws Exception {
+        assertEquals(1, receivedMessages.size());
+        logger.info(String.format("Schema message: `%s`", receivedMessages.get(0).getBody()));
+        SchemaPubSubMessage actual = AwsTestUtils.unwrapSingleFirst(receivedMessages, SchemaPubSubMessage.class);
+        assertEquals(expected.getKind(), actual.getKind());
+        assertEquals(expected.getOp(), actual.getOp());
+    }
+
+    private ClientResponse sendRequest(String url, String method, boolean addHeaders, Object body, MediaType contentType) {
+        WebResource resource = restClient.resource(url);
+        WebResource.Builder resourceBuilder = resource.getRequestBuilder();
+        if (contentType != null) {
+            resourceBuilder.accept(contentType);
+            resourceBuilder.type(contentType);
+        }
+        if (addHeaders) {
+            dpsHeaders.getHeaders().forEach(resourceBuilder::header);
+        }
+        return (body == null) ? resourceBuilder.method(method, ClientResponse.class) :  resourceBuilder.method(method, ClientResponse.class, body);
+    }
+}

testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/StorageTestUtils.java

@@ -37,25 +37,24 @@ public class StorageTestUtils {
 
     private final IStorageService storageService;
     private final HashSet<OperationType> allowedOps = new HashSet<>();
+    private final String recordKind;
 
-    public StorageTestUtils(DpsHeaders headers, HttpResponseBodyMapper mapper) {
+    public StorageTestUtils(DpsHeaders headers, HttpResponseBodyMapper mapper, SchemaTestUtils schemaTestUtils) {
         StorageAPIConfig config = StorageAPIConfig.builder().rootUrl(Config.Instance().StorageServicePath).build();
         StorageFactory subscriptionFactory = new StorageFactory(config, mapper);
         storageService = subscriptionFactory.create(headers);
         allowedOps.add(OperationType.create);
         allowedOps.add(OperationType.delete);
+        recordKind = schemaTestUtils.createSchema();
     }
 
-    protected static final String PERSISTABLE_REFERENCE = "%7B%22LB_CRS%22%3A%22%257B%2522WKT%2522%253A%2522PROJCS%255B%255C%2522British_National_Grid%255C%2522%252CGEOGCS%255B%255C%2522GCS_OSGB_1936%255C%2522%252CDATUM%255B%255C%2522D_OSGB_1936%255C%2522%252CSPHEROID%255B%255C%2522Airy_1830%255C%2522%252C6377563.396%252C299.3249646%255D%255D%252CPRIMEM%255B%255C%2522Greenwich%255C%2522%252C0.0%255D%252CUNIT%255B%255C%2522Degree%255C%2522%252C0.0174532925199433%255D%255D%252CPROJECTION%255B%255C%2522Transverse_Mercator%255C%2522%255D%252CPARAMETER%255B%255C%2522False_Easting%255C%2522%252C400000.0%255D%252CPARAMETER%255B%255C%2522False_Northing%255C%2522%252C-100000.0%255D%252CPARAMETER%255B%255C%2522Central_Meridian%255C%2522%252C-2.0%255D%252CPARAMETER%255B%255C%2522Scale_Factor%255C%2522%252C0.9996012717%255D%252CPARAMETER%255B%255C%2522Latitude_Of_Origin%255C%2522%252C49.0%255D%252CUNIT%255B%255C%2522Meter%255C%2522%252C1.0%255D%252CAUTHORITY%255B%255C%2522EPSG%255C%2522%252C27700%255D%255D%2522%252C%2522Type%2522%253A%2522LBCRS%2522%252C%2522EngineVersion%2522%253A%2522PE_10_3_1%2522%252C%2522AuthorityCode%2522%253A%257B%2522Authority%2522%253A%2522EPSG%2522%252C%2522Code%2522%253A%252227700%2522%257D%252C%2522Name%2522%253A%2522British_National_Grid%2522%257D%22%2C%22TRF%22%3A%22%257B%2522WKT%2522%253A%2522GEOGTRAN%255B%255C%2522OSGB_1936_To_WGS_1984_Petroleum%255C%2522%252CGEOGCS%255B%255C%2522GCS_OSGB_1936%255C%2522%252CDATUM%255B%255C%2522D_OSGB_1936%255C%2522%252CSPHEROID%255B%255C%2522Airy_1830%255C%2522%252C6377563.396%252C299.3249646%255D%255D%252CPRIMEM%255B%255C%2522Greenwich%255C%2522%252C0.0%255D%252CUNIT%255B%255C%2522Degree%255C%2522%252C0.0174532925199433%255D%255D%252CGEOGCS%255B%255C%2522GCS_WGS_1984%255C%2522%252CDATUM%255B%255C%2522D_WGS_1984%255C%2522%252CSPHEROID%255B%255C%2522WGS_1984%255C%2522%252C6378137.0%252C298.257223563%255D%255D%252CPRIMEM%255B%255C%2522Greenwich%255C%2522%252C0.0%255D%252CUNIT%255B%255C%2522Degree%255C%2522%252C0.0174532925199433%255D%255D%252CMETHOD%255B%255C%2522Position_Vector%255C%2522%255D%252CPARAMETER%255B%255C%2522X_Axis_Translation%255C%2522%252C446.448%255D%252CPARAMETER%255B%255C%2522Y_Axis_Translation%255C%2522%252C-125.157%255D%252CPARAMETER%255B%255C%2522Z_Axis_Translation%255C%2522%252C542.06%255D%252CPARAMETER%255B%255C%2522X_Axis_Rotation%255C%2522%252C0.15%255D%252CPARAMETER%255B%255C%2522Y_Axis_Rotation%255C%2522%252C0.247%255D%252CPARAMETER%255B%255C%2522Z_Axis_Rotation%255C%2522%252C0.842%255D%252CPARAMETER%255B%255C%2522Scale_Difference%255C%2522%252C-20.489%255D%252CAUTHORITY%255B%255C%2522EPSG%255C%2522%252C1314%255D%255D%2522%252C%2522Type%2522%253A%2522STRF%2522%252C%2522EngineVersion%2522%253A%2522PE_10_3_1%2522%252C%2522AuthorityCode%2522%253A%257B%2522Authority%2522%253A%2522EPSG%2522%252C%2522Code%2522%253A%25221314%2522%257D%252C%2522Name%2522%253A%2522OSGB_1936_To_WGS_1984_Petroleum%2522%257D%22%2C%22Type%22%3A%22EBCRS%22%2C%22EngineVersion%22%3A%22PE_10_3_1%22%2C%22Name%22%3A%22OSGB+1936+*+UKOOA-Pet+%2F+British+National+Grid+%5B27700%2C1314%5D%22%2C%22AuthorityCode%22%3A%7B%22Authority%22%3A%22MyCompany%22%2C%22Code%22%3A%2227700006%22%7D%7D";
     private static final String RECORD_ID_PREFIX_TEST = TenantUtils.getTenantName(true) + ":query:";
-    private static final long NOW = System.currentTimeMillis();
-    private static final String KIND_TEST = TenantUtils.getTenantName(true) + ":ds:query:1.0." + NOW;
     public PubSubInfo createAndPutRecord(String correlationId, String legalTag) throws StorageException {
         String recordId = RECORD_ID_PREFIX_TEST + correlationId;
-        Record[] records = RecordUtils.createRecordsWithReference(1, recordId, KIND_TEST, legalTag, PERSISTABLE_REFERENCE, "CRS", true);
+        Record[] records = RecordUtils.createRecordsWithReference(1, recordId, recordKind, legalTag, true);
         storageService.upsertRecord(records);
         // We append the index in RecordUtils to ensure the IDs are unique. Maybe we don't need to do that.
-        return PubSubInfo.builder().id(records[0].getId()).kind(KIND_TEST).op(OperationType.create).build();
+        return PubSubInfo.builder().id(records[0].getId()).kind(recordKind).op(OperationType.create).build();
     }
 
     public void assertFirstMessagesSimilar(PubSubInfo expected, List<Message> receivedMessages) throws Exception {

testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TestNotificationsEndpoint.java

@@ -38,6 +38,7 @@ import org.opengroup.osdu.core.common.notification.ISubscriptionService;
 import org.opengroup.osdu.core.common.notification.SubscriptionAPIConfig;
 import org.opengroup.osdu.core.common.notification.SubscriptionException;
 import org.opengroup.osdu.core.common.notification.SubscriptionFactory;
+import org.opengroup.osdu.notification.subscriptions.SchemaTestUtils.SchemaPubSubMessage;
 import org.opengroup.osdu.notification.util.*;
 
 import java.time.Instant;
@@ -53,6 +54,7 @@ import java.util.UUID;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 public class TestNotificationsEndpoint {
@@ -61,8 +63,10 @@ public class TestNotificationsEndpoint {
     private static StorageTestUtils storageUtils;
     private static LegalTagUtils legalUtils;
     private static FileTestUtils fileUtils;
+    private static SchemaTestUtils schemaTestUtils;
 
     private static final int FILE_QUEUE_NUM = 0;
+    private static final int SCHEMA_QUEUE_NUM = 0;
     private static final int INDEXER_STORAGE_INDEXER_QUEUE_NUM = 0;
     private static final int INDEXER_STORAGE_STORAGE_QUEUE_NUM = 1;
     private static final int LEGAL_STORAGE_STORAGE_QUEUE_NUM = 0;
@@ -74,6 +78,7 @@ public class TestNotificationsEndpoint {
     private static final String INDEXER_SERVICE_NAME = "indexer";
     private static final String LEGAL_SERVICE_NAME = "legal";
     private static final String STORAGE_SERVICE_NAME = "storage";
+    private static final String SCHEMA_SERVICE_NAME = "schema";
     private final Object mutex = new Object();
     private static final String FILE_LEGAL_TAG_NAME = String.format("%s-file-testing-%d", TenantUtils.getTenantName(true), System.currentTimeMillis());
     private static final String INDEXER_LEGAL_TAG_NAME = String.format("%s-indexer-testing-%d", TenantUtils.getTenantName(true), System.currentTimeMillis());
@@ -217,7 +222,8 @@ public class TestNotificationsEndpoint {
 
         DpsHeaders dpsHeaders = testUtils.getDpsHeaders();
         // Set up the legal service used by this test suite
-        storageUtils = new StorageTestUtils(dpsHeaders, responseBodyMapper);
+        schemaTestUtils = new SchemaTestUtils(dpsHeaders);
+        storageUtils = new StorageTestUtils(dpsHeaders, responseBodyMapper, schemaTestUtils);
         legalUtils = new LegalTagUtils(dpsHeaders, responseBodyMapper);
         subscriptionService = subscriptionFactory.create(dpsHeaders);
         fileUtils = new FileTestUtils(dpsHeaders);
@@ -261,7 +267,7 @@ public class TestNotificationsEndpoint {
     private static void getTopicNames() {
         // Get the OSDU topics for each of the services that emit notifications.
         String instanceName = getEnv("OSDU_INSTANCE_NAME");
-        String[] services = new String[] {FILE_SERVICE_NAME, INDEXER_SERVICE_NAME, LEGAL_SERVICE_NAME, STORAGE_SERVICE_NAME};
+        String[] services = new String[] {FILE_SERVICE_NAME, INDEXER_SERVICE_NAME, LEGAL_SERVICE_NAME, STORAGE_SERVICE_NAME, SCHEMA_SERVICE_NAME};
         List<String> ssmNames = new ArrayList<>(services.length);
         Map<String, String> parameterReverseLookup = new HashMap<>(services.length);
         Arrays.stream(services).forEach(serviceName -> {
@@ -351,7 +357,24 @@ public class TestNotificationsEndpoint {
     }
 
     @Test
-    @Ignore("On the pipeline, this integration test can take way more than 10 minutes, which is determined to be unrealistic.")
+    public void test_Schema_Notifications_Received() throws Exception {
+        synchronized (mutex) {
+            PushQueue schemaQueue = queues.get(SCHEMA_QUEUE_NUM);
+            try {
+                schemaQueue.register(SCHEMA_SERVICE_NAME);
+                SchemaPubSubMessage expected = schemaTestUtils.createSchemaForTest();
+                List<List<Message>> messagesLists = pollForMessages(schemaQueue);
+                assertEquals(1, messagesLists.size());
+                schemaTestUtils.validate(expected, messagesLists.get(0));
+            } catch (Exception e) {
+                System.err.println(e.toString());
+                e.printStackTrace();
+                throw e;
+            }
+        }
+    }
+
+    @Test
     public void test_Indexer_Notifications_Received() throws Exception {
         synchronized (mutex) {
             PushQueue storageQueue = queues.get(INDEXER_STORAGE_STORAGE_QUEUE_NUM);
@@ -377,8 +400,8 @@ public class TestNotificationsEndpoint {
                 assertEquals(expectedStorageMessage.getKind(), indexerMessage.kind());
                 assertEquals(expectedStorageMessage.getId(), indexerMessage.id());
                 assertEquals("create", indexerMessage.operationType());
-                assertEquals("WARN", indexerMessage.status());
-                assertNotEquals(null, indexerMessage.indexProgress());
+                assertEquals("SUCCESS", indexerMessage.status());
+                assertNotNull(indexerMessage.indexProgress());
                 assertEquals(200, indexerMessage.indexProgress().statusCode());
                 Instant lastChanged = Instant.parse(indexerMessage.indexProgress().lastUpdateTime());
                 assertTrue(lastChanged.isAfter(start) && lastChanged.isBefore(end));