diff --git a/provider/notification-aws/build-aws/buildspec.yaml b/provider/notification-aws/build-aws/buildspec.yaml
index f9e607cf66c9fe966f36ad6fb495d8c5d84880ed..cce51eb61d9030e516f22fb253740a823899c368 100644
--- a/provider/notification-aws/build-aws/buildspec.yaml
+++ b/provider/notification-aws/build-aws/buildspec.yaml
@@ -47,18 +47,19 @@ phases:
pre_build:
commands:
- echo "Logging in to Amazon ECR..."
- - aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin ${ECR_REGISTRY}
+ - aws ecr get-login-password --region ${AWS_REGION} | docker login --username AWS --password-stdin ${ECR_REGISTRY} # authenticate with ECR via the AWS CLI
- echo "Logging into Docker Hub..."
- docker login -u ${DOCKER_USERNAME} -p ${DOCKER_PASSWORD}
-
build:
commands:
- export REPO_NAME=${PWD##*/}
- export OUTPUT_DIR="dist"
- export BRANCH_NAME=`echo ${CODEBUILD_SOURCE_VERSION} | awk '{gsub("refs/heads/","");gsub("\\.","-");gsub("[[:space:]]","-")}1' | sed 's/\//-/g' | awk '{print tolower($0)}'`
- export ECR_TAG=`echo build.${BRANCH_NAME}.${CODEBUILD_BUILD_NUMBER}.${CODEBUILD_RESOLVED_SOURCE_VERSION} | cut -c 1-120`
+ - export PUSH_ENDPOINT_ECR_TAG=`echo push_endpoint.${BRANCH_NAME}.${CODEBUILD_BUILD_NUMBER}.${CODEBUILD_RESOLVED_SOURCE_VERSION} | cut -c 1-120`
- export ECR_IMAGE=${ECR_REGISTRY}:${ECR_TAG}
- export ECR_IMAGE_BRANCH_LATEST=${ECR_REGISTRY}:${BRANCH_NAME}
+ - export PUSH_ENDPOINT_IMAGE=${ECR_REGISTRY}:${PUSH_ENDPOINT_ECR_TAG}
- export INTEGRATION_TEST_OUTPUT=${OUTPUT_DIR}/testing/integration
- export INTEGRATION_TEST_OUTPUT_BIN=${INTEGRATION_TEST_OUTPUT}/bin
- mkdir -p ${OUTPUT_DIR}/bin
diff --git a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationQueueService.java b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationQueueService.java
index 40f21f3eaee9707649605828ba405e78e28e6799..0b4d818faed050cecf7fbd0852fa0158d899d991 100644
--- a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationQueueService.java
+++ b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationQueueService.java
@@ -1,16 +1,17 @@
-// Copyright © Amazon Web Services
-//
-// 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.
+/* 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.provider.aws.queue;
diff --git a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetryQueueService.java b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetryQueueService.java
index bc7d33c14f9379406615a57f33c5836cbc0193f2..55f89b91fccec2be10d54f87f44e0c5435662f2e 100644
--- a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetryQueueService.java
+++ b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetryQueueService.java
@@ -1,16 +1,18 @@
-// Copyright © Amazon Web Services
-//
-// 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.
+/* 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.provider.aws.queue;
import com.amazonaws.services.sqs.model.Message;
diff --git a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetrySQSHandler.java b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetrySQSHandler.java
index 00f68fcf110d56813cacfd48207638d655d80c68..6aaf28899a29493cd26b26a3f33b00085509d437 100644
--- a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetrySQSHandler.java
+++ b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationRetrySQSHandler.java
@@ -1,16 +1,18 @@
-// Copyright © Amazon Web Services
-//
-// 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.
+/* 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.provider.aws.queue;
import com.amazonaws.services.sqs.AmazonSQS;
diff --git a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationSQSHandler.java b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationSQSHandler.java
index 7a1078e858fc4e96f4fb4136991b625f857a39b6..d4e7fe89e8588fcfa709a7990d3d430f6ce5c6d3 100644
--- a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationSQSHandler.java
+++ b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/NotificationSQSHandler.java
@@ -1,16 +1,18 @@
-// Copyright © Amazon Web Services
-//
-// 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.
+/* 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.provider.aws.queue;
import com.amazonaws.services.sqs.AmazonSQS;
diff --git a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationQueueServiceImpl.java b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationQueueServiceImpl.java
index da20bf30a1f4e627a14fb438d4bed1067d8683d7..2c0d79d776b79015a73057f2b9dcd63879bb4e44 100644
--- a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationQueueServiceImpl.java
+++ b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationQueueServiceImpl.java
@@ -1,16 +1,18 @@
-// Copyright © Amazon Web Services
-//
-// 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.
+/* 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.provider.aws.queue.impl;
import com.amazonaws.services.dynamodbv2.datamodeling.PaginatedQueryList;
@@ -115,6 +117,7 @@ public class NotificationQueueServiceImpl implements NotificationQueueService {
}
public void processMessagesBySubscription(Subscription subscription, List<Message> messages, String dataPartitionId) {
+ logger.info("Processing {} messages for subscription {}", messages.size(), subscription.getId());
List<Message> messagesToRetry = new ArrayList<>();
DynamoDBQueryHelperV2 dynamoDBQueryHelper = dynamoDBQueryHelperFactory.getQueryHelperUsingSSM(failedNotificationTablePath);
try {
@@ -122,13 +125,15 @@ public class NotificationQueueServiceImpl implements NotificationQueueService {
gsiQuery.setPartitionIdSubscriptionId(String.join("-", dataPartitionId, subscription.getId()));
PaginatedQueryList<FailedNotificationDoc> results = dynamoDBQueryHelper.queryByGSI(FailedNotificationDoc.class, gsiQuery);
+ logger.info("Failed notifications result: {}", results);
if (results != null && !results.isEmpty()) {
- logger.debug("Subscription {} has previous failed messages", subscription.getId());
+ logger.info("Subscription {} has previous failed messages", subscription.getId());
messagesToRetry.addAll(messages);
} else {
+ logger.info("Processing messages for subscription that has not yet failed: {}", subscription.getPushEndpoint());
boolean hasPreviousFailed = false;
for (Message message : messages) {
- logger.debug("Processing Notification for messageId: {}, Subscription Endpoint: {}", message.getMessageId(), subscription.getPushEndpoint());
+ logger.info("Processing Notification for messageId: {}, Subscription Endpoint: {}", message.getMessageId(), subscription.getPushEndpoint());
if (hasPreviousFailed) {
messagesToRetry.add(message);
continue;
@@ -145,12 +150,13 @@ public class NotificationQueueServiceImpl implements NotificationQueueService {
logger.error("Exception in processMessagesBySubscription for {} :", subscription.getPushEndpoint(), e);
messagesToRetry.addAll(messages);
}
- logger.info("Messages retry size :: {}", messagesToRetry.size());
+ logger.info("Messages retry size :: {} Subscription ID: {}", messagesToRetry.size(), subscription.getId());
messagesToRetry.forEach(message -> insertFailedNotification(message, subscription, dataPartitionId, dynamoDBQueryHelper));
}
private boolean notifySubscriber(Subscription subscription, String messageBody, Map<String, String> headerAttributes) {
// Only process this subscription if the topics match.
+ logger.debug("Subscription: {}, Message Body: {}, Message Headers: {}", subscription, messageBody, headerAttributes);
if (!subscription.getTopic().equals(headerAttributes.get(PublishRequestBuilder.OSDU_TOPIC_ATTRIBUTE_NAME))) return true;
HttpResponse response;
diff --git a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationRetryQueueServiceImpl.java b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationRetryQueueServiceImpl.java
index b51df1790df1da3b6f1b66ea27c5895bc4fcd323..de55ca71ce6487d40469fbe3ebae69116d53ff0b 100644
--- a/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationRetryQueueServiceImpl.java
+++ b/provider/notification-aws/src/main/java/org/opengroup/osdu/notification/provider/aws/queue/impl/NotificationRetryQueueServiceImpl.java
@@ -1,16 +1,18 @@
-// Copyright © Amazon Web Services
-//
-// 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.
+/* 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.provider.aws.queue.impl;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBDeleteExpression;
@@ -101,12 +103,13 @@ public class NotificationRetryQueueServiceImpl implements NotificationRetryQueue
DynamoDBQueryHelperV2 dynamoDBQueryHelper = dynamoDBQueryHelperFactory.getQueryHelperUsingSSM(failedNotificationTablePath);
boolean hasPreviousFailed = false;
List<RetryProcessResult> notificationResults = new ArrayList<>();
+ logger.info("Retrying {} messages from the Retry Queue", messages.size());
for (Message message : messages) {
RetryProcessResult processResult = new RetryProcessResult();
processResult.setMessage(message);
- logger.debug("Processing Notification for messageId: {}, Subscription Endpoint: {}", message.getMessageId(), subscription.getPushEndpoint());
+ logger.info("Processing Notification for messageId: {}, Subscription Endpoint: {}", message.getMessageId(), subscription.getPushEndpoint());
if (hasPreviousFailed) {
processResult.setResult(NotificationResult.NACK);
notificationResults.add(processResult);
@@ -119,6 +122,7 @@ public class NotificationRetryQueueServiceImpl implements NotificationRetryQueue
if (!response.isSuccessCode()) {
processResult.setResult(NotificationResult.NACK);
hasPreviousFailed = true;
+ logger.error("Failed to process messageId {} for subscriber {}. Error message: {}", message.getMessageId(), subscription.getId(), response.getBody());
} else {
processResult.setResult(NotificationResult.ACK);
FailedNotificationDoc doc = dynamoDBQueryHelper.loadByPrimaryKey(FailedNotificationDoc.class,
diff --git a/testing/notification-test-aws/build-aws/prepare-dist.sh b/testing/notification-test-aws/build-aws/prepare-dist.sh
index 3aaf33a954f52dae47001d8f4da7c9449860e7c4..a82740f3e97968bf7cdfcda58863587ef120288e 100755
--- a/testing/notification-test-aws/build-aws/prepare-dist.sh
+++ b/testing/notification-test-aws/build-aws/prepare-dist.sh
@@ -35,6 +35,10 @@ echo $OUTPUT_DIR
echo $INTEGRATION_TEST_OUTPUT_DIR
echo $INTEGRATION_TEST_OUTPUT_BIN_DIR
+# Build the Push Endpoint image
+(cd $INTEGRATION_TEST_SOURCE_DIR_AWS/build-aws/push-endpoint && mvn -B -ntp -s ./maven/settings.xml package && docker build -t $PUSH_ENDPOINT_IMAGE -f Dockerfile --no-cache . && docker push $PUSH_ENDPOINT_IMAGE)
+# Finished building the Push Endpoint image
+
rm -rf "$INTEGRATION_TEST_OUTPUT_DIR"
mkdir -p "$INTEGRATION_TEST_OUTPUT_DIR" && mkdir -p "$INTEGRATION_TEST_OUTPUT_BIN_DIR"
echo "Building integration testing assemblies and gathering artifacts..."
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/verify_register-logs.sh b/testing/notification-test-aws/build-aws/push-endpoint/Dockerfile
old mode 100755
new mode 100644
similarity index 58%
rename from testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/verify_register-logs.sh
rename to testing/notification-test-aws/build-aws/push-endpoint/Dockerfile
index b16929e57968f1022ad8c7fa413a5623a9e6211c..48b47e5a79f66b6f2d47b430cc26175c5aa64dab
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/verify_register-logs.sh
+++ b/testing/notification-test-aws/build-aws/push-endpoint/Dockerfile
@@ -12,12 +12,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-CORE_K8S_NAMESPACE="$AWS_SERVICE_NAMESPACE-core"
-register_pod=$(kubectl get pods -n $CORE_K8S_NAMESPACE | grep "Running" | grep "os-register" | awk '{ print $1 }')
-register_logs=$(kubectl logs $register_pod -n $CORE_K8S_NAMESPACE --since=2m)
+# https://docs.spring.io/spring-boot/docs/current/reference/html/deployment.html
+FROM amazoncorretto:17
-if echo $register_logs | grep $1; then
- exit 0
-else
- exit 1
-fi
+ARG JAR_FILE=target/*spring-boot.jar
+# Harcoding this value since Notification-core requires this variable. AWS does not use it. Might change in future
+ENV ENVIRONMENT=DEV
+
+
+WORKDIR /
+COPY ${JAR_FILE} app.jar
+COPY /entrypoint.sh /entrypoint.sh
+EXPOSE 8080
+
+ENTRYPOINT ["/bin/sh", "-c", ". /entrypoint.sh"]
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/entrypoint.sh b/testing/notification-test-aws/build-aws/push-endpoint/entrypoint.sh
new file mode 100644
index 0000000000000000000000000000000000000000..ad36924a8a44f39172321142486daf67eeb75c09
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/entrypoint.sh
@@ -0,0 +1,16 @@
+# 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.
+
+
+java $JAVA_OPTS -jar /app.jar
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/maven/settings.xml b/testing/notification-test-aws/build-aws/push-endpoint/maven/settings.xml
new file mode 100644
index 0000000000000000000000000000000000000000..274a34febd0abf7379555dc4ee10e1d9c64ee67f
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/maven/settings.xml
@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+
+<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd">
+
+ <profiles>
+ <profile>
+ <id>aws-osdu-dev-maven</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <repositories>
+ <repository>
+ <id>aws-osdu-dev-maven</id>
+ <url>${env.AWS_OSDU_DEV_MAVEN_URL}</url>
+ </repository>
+ <repository>
+ <id>gitlab-os-core-common-maven</id>
+ <url>https://community.opengroup.org/api/v4/projects/67/packages/maven</url>
+ </repository>
+ <repository>
+ <id>gitlab-os-core-lib-aws-maven</id>
+ <url>https://community.opengroup.org/api/v4/projects/68/packages/maven</url>
+ </repository>
+ </repositories>
+ </profile>
+ <profile>
+ <id>credentialsConfiguration</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <properties>
+ <deployment.environment>dev</deployment.environment>
+ <aws.accessKeyId>no-default</aws.accessKeyId>
+ <aws.secretKey>no-default</aws.secretKey>
+ <azure.devops.username>Another-Access-Token-2021</azure.devops.username>
+ <azure.devops.token>no-default</azure.devops.token>
+ </properties>
+ </profile>
+ <profile>
+ <id>sonar</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <properties>
+ <sonar.host.url>
+ ${env.SONAR_URL}
+ </sonar.host.url>
+ </properties>
+ </profile>
+ </profiles>
+
+ <servers>
+ <server>
+ <id>aws-osdu-dev-maven</id>
+ <username>aws</username>
+ <password>${env.AWS_OSDU_DEV_MAVEN_AUTH_TOKEN}</password>
+ </server>
+ </servers>
+
+ <mirrors>
+ <mirror>
+ <id>aws-osdu-dev-maven</id>
+ <name>aws-osdu-dev-maven</name>
+ <url>${env.AWS_OSDU_DEV_MAVEN_URL}</url>
+ <mirrorOf>!gitlab-os-core-common-maven,!gitlab-os-core-lib-aws-maven</mirrorOf>
+ </mirror>
+ </mirrors>
+
+ <activeProfiles>
+ <activeProfile>credentialsConfiguration</activeProfile>
+ </activeProfiles>
+
+</settings>
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/pom.xml b/testing/notification-test-aws/build-aws/push-endpoint/pom.xml
new file mode 100644
index 0000000000000000000000000000000000000000..0c8d0b894a4f297e0c65e478dd1a34a00b825ca4
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/pom.xml
@@ -0,0 +1,142 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.example</groupId>
+ <artifactId>notification-push-endpoint</artifactId>
+ <packaging>jar</packaging>
+ <version>1.0-SNAPSHOT</version>
+
+ <parent>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-parent</artifactId>
+ <version>3.1.4</version>
+ <relativePath/>
+ </parent>
+
+ <properties>
+ <maven.compiler.source>17</maven.compiler.source>
+ <maven.compiler.target>17</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <os-core-common.version>0.23.0-SNAPSHOT</os-core-common.version>
+ <spring-boot-maven-plugin.version>2.7.6</spring-boot-maven-plugin.version>
+ <aws-java-sdk.version>1.11.1018</aws-java-sdk.version>
+ <aws-encryption-sdk-java.version>2.4.0</aws-encryption-sdk-java.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-security</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-web</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opengroup.osdu</groupId>
+ <artifactId>os-core-common</artifactId>
+ <version>${os-core-common.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-encryption-sdk-java</artifactId>
+ <version>${aws-encryption-sdk-java.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-sqs</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-sts</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
+ </dependencies>
+
+ <profiles>
+ <profile>
+ <id>aws-dev</id>
+ <activation>
+ <property>
+ <name>deployment.environment</name>
+ <value>dev</value>
+ </property>
+ </activation>
+ <properties>
+ <aws.version>1.11.1018</aws.version>
+ </properties>
+ </profile>
+ </profiles>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <version>${spring-boot-maven-plugin.version}</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>repackage</goal>
+ </goals>
+ <configuration>
+ <classifier>spring-boot</classifier>
+ <mainClass>
+ org.example.Application
+ </mainClass>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.5.1</version>
+ <configuration>
+ <createDependencyReducedPom>false</createDependencyReducedPom>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
\ No newline at end of file
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/setup_teardown_endpoint.py b/testing/notification-test-aws/build-aws/push-endpoint/setup_teardown_endpoint.py
new file mode 100644
index 0000000000000000000000000000000000000000..959f85e3b8f6c23b2f7dae5bf469f9a8bc9d8455
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/setup_teardown_endpoint.py
@@ -0,0 +1,572 @@
+# 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.
+
+import asyncio
+import json
+import subprocess as sp
+import argparse as ap
+import os
+import time
+
+import boto3
+import typing
+
+NOTIFICATION_SUFFIX = "push-endpoint"
+
+
+SERVICE_ROLE_ASSUME_POLICY_TEMPLATE = """{{
+ "Version": "2012-10-17",
+ "Statement": [
+ {{
+ "Effect": "Allow",
+ "Principal": {{
+ "Federated": "arn:aws:iam::{0}:oidc-provider/{1}"
+ }},
+ "Action": "sts:AssumeRoleWithWebIdentity",
+ "Condition": {{
+ "StringEquals": {{
+ "{1}:sub": "system:serviceaccount:{2}:{3}"
+ }}
+ }}
+ }}
+ ]
+}}"""
+
+
+SERVICE_ROLE_PERMISSIONS_POLICY_TEMPLATE = """{{
+ "Version": "2012-10-17",
+ "Statement": [
+ {{
+ "Effect": "Allow",
+ "Action": [
+ "sqs:SendMessage"
+ ],
+ "Resource": ["{0}"]
+ }}
+ ]
+}}"""
+
+
+QUEUE_URL_TEMPLATE = """
+ - name: QUEUE_NUM_{0}
+ value: {1}"""
+
+ALLOWED_ORIGINS_TEMPLATE = """
+ - prefix: {0}"""
+
+SERVICE_ACCOUNT_NAME = "push-endpoint-sa"
+PUSH_ENDPOINT_POD_NAME = "push-endpoint"
+PUSH_ENDPOINT_PATH = "/api/push-endpoint"
+PUSH_ENDPOINT_K8S_TEMPLATE = """
+apiVersion: v1
+kind: Namespace
+metadata:
+ labels:
+ istio-injection: enabled
+ name: {push_namespace}
+spec:
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+ annotations:
+ eks.amazonaws.com/role-arn: {role_arn}
+ name: {name}
+ namespace: {push_namespace}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ labels:
+ service: {name}
+ name: {name}
+ namespace: {push_namespace}
+spec:
+ ports:
+ - name: http
+ port: 8080
+ protocol: TCP
+ targetPort: 8080
+ selector:
+ app.kubernetes.io/instance: {name}
+ app.kubernetes.io/name: {name}
+ type: ClusterIP
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ labels:
+ app.kubernetes.io/instance: {name}
+ app.kubernetes.io/name: {name}
+ name: {name}
+ namespace: {push_namespace}
+spec:
+ selector:
+ matchLabels:
+ app.kubernetes.io/instance: {name}
+ app.kubernetes.io/name: {name}
+ template:
+ metadata:
+ annotations: null
+ labels:
+ app.kubernetes.io/instance: {name}
+ app.kubernetes.io/name: {name}
+ spec:
+ automountServiceAccountToken: true
+ containers:
+ - env:
+ - name: HMAC_SECRET
+ value: "{hmac_secret}"
+ - name: AWS_REGION
+ value: "{aws_region}"
+ - name: MAX_QUEUE_NUM
+ value: "{num_queues}"
+{queue_urls}
+ image: "{image}"
+ imagePullPolicy: Always
+ name: {name}
+ ports:
+ - containerPort: 8080
+ name: http
+ protocol: TCP
+ resources:
+ limits:
+ memory: 900M
+ requests:
+ cpu: 1000m
+ memory: 900M
+ securityContext:
+ allowPrivilegeEscalation: false
+ capabilities:
+ drop:
+ - ALL
+ readOnlyRootFilesystem: true
+ runAsNonRoot: true
+ runAsUser: 10001
+ volumeMounts:
+ - mountPath: /tmp
+ name: tmp-volume
+ securityContext:
+ fsGroup: 1337
+ seccompProfile:
+ type: RuntimeDefault
+ serviceAccountName: {name}
+ volumes:
+ - emptyDir:
+ sizeLimit: 2Gi
+ name: tmp-volume
+---
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+ name: {name}-allow-services
+ namespace: {push_namespace}
+spec:
+ action: ALLOW
+ rules:
+ - from:
+ - source:
+ principals:
+ - cluster.local/ns/{ingress_namespace}/sa/{ingress_gateway_name}
+ - cluster.local/ns/{core_namespace}/sa/os-notification
+ - cluster.local/ns/{core_namespace}/sa/os-register
+ selector:
+ matchLabels:
+ app.kubernetes.io/name: {name}
+---
+apiVersion: networking.istio.io/v1alpha3
+kind: DestinationRule
+metadata:
+ name: {name}
+ namespace: {push_namespace}
+spec:
+ host: {name}
+ trafficPolicy:
+ portLevelSettings:
+ - connectionPool:
+ http:
+ http1MaxPendingRequests: 10000
+ http2MaxRequests: 100
+ tcp:
+ maxConnections: 0
+ port:
+ number: 8080
+ tls:
+ mode: ISTIO_MUTUAL
+---
+apiVersion: networking.istio.io/v1alpha3
+kind: VirtualService
+metadata:
+ name: {name}
+ namespace: {push_namespace}
+spec:
+ gateways:
+ - {ingress_namespace}/{osdu_gateway_name}
+ hosts:
+ - '*'
+ http:
+ - corsPolicy:
+ allowCredentials: true
+ allowHeaders:
+ - Authorization
+ - Data-Partition-Id
+ - Correlation-Id
+ - Content-Type
+ allowMethods:
+ - POST
+ - GET
+ - PATCH
+ - PUT
+ - DELETE
+ allowOrigins:
+{allowed_origins}
+ maxAge: 60m
+ headers:
+ response:
+ remove:
+ - access-control-allow-origin
+ - access-control-allow-credentials
+ set:
+ access-control-allow-headers: Authorization, Data-Partition-Id, Correlation-Id,
+ Content-Type
+ access-control-allow-methods: POST, GET, PATCH, PUT, DELETE
+ match:
+ - uri:
+ prefix: {api_path}
+ route:
+ - destination:
+ host: {name}
+ port:
+ number: 8080
+"""
+
+class CommonParameters:
+ def __init__(self, osdu_instance: str, eks_cluster: str, region: str,
+ num_queues: int, env_file: str, base_url: str, domain_name: str,
+ endpoint_suffix: str, kube_file: typing.Optional[str], api_path: str):
+ self.account_id = boto3.client('sts').get_caller_identity().get('Account')
+ self.osdu_instance = osdu_instance
+ self.eks_cluster = eks_cluster
+ self.region = region
+ self.domain_name = domain_name
+ self.endpoint_suffix = endpoint_suffix
+ self.kube_file = kube_file
+ self.namespace_prefix: str = ""
+ self.ingress_namespace: str = ""
+ self.ingress_gateway: str = ""
+ self.osdu_gateway: str = ""
+ self.amp_workspace: str = ""
+ self.num_queues = num_queues
+ self.env_file = env_file
+ self.base_url = base_url
+ self.sa_role_policy_name = "int-test-notification-policy"
+ self._set_parameters()
+ self.api_path = api_path
+ self.wildcard_path = f"{self.api_path}/*"
+ self.sqs_client = boto3.client('sqs', region_name=region)
+ self.iam_client = boto3.client('iam', region_name=region)
+
+ if kube_file is None:
+ self.kube_file = f"{self.push_namespace}.yaml"
+
+ def _set_parameters(self):
+ sns_client = boto3.client('ssm')
+ parameters = {
+ f"/osdu/eks/{self.eks_cluster}/amp/workspace/id": "amp_workspace",
+ f"/osdu/eks/{self.eks_cluster}/instances/{self.osdu_instance}/ingress/osdu-gateway/name": "osdu_gateway",
+ f"/osdu/eks/{self.eks_cluster}/instances/{self.osdu_instance}/ingress/ingress-gateway/name": "ingress_gateway",
+ f"/osdu/eks/{self.eks_cluster}/instances/{self.osdu_instance}/ingress/namespace": "ingress_namespace",
+ f"/osdu/instances/{self.osdu_instance}/config/k8s/namespace-prefix": "namespace_prefix",
+ }
+ params = sns_client.get_parameters(Names=[key for key in parameters])
+ for parameter in params["Parameters"]:
+ parameter_path = parameter["Name"]
+ short_name = parameters[parameter_path]
+ setattr(self, short_name, parameter["Value"])
+
+ self.core_namespace = self.get_namespace("core")
+ self.push_namespace = self.get_namespace(self.endpoint_suffix)
+ print("Push Endpoint namespace:", self.push_namespace)
+
+ def get_namespace(self, suffix: str) -> str:
+ return f"{self.namespace_prefix}-{suffix}"
+
+ def get_service_role(self) -> str:
+ return f"{self.region}-{self.push_namespace}"
+
+ def get_sqs_queue_name(self, queue_num: int) -> str:
+ return f"{self.push_namespace}-{queue_num}"
+
+ def get_oidc_id(self) -> str:
+ eks_client = boto3.client('eks', region_name=self.region)
+ cluster_info = eks_client.describe_cluster(name=self.eks_cluster)
+ return cluster_info["cluster"]["identity"]["oidc"]["issuer"] \
+ [len("https://"):]
+
+ def create_policy_document(self, queue_arns: typing.List[str]) -> str:
+ service_role_name = self.get_service_role()
+ oidc_id = self.get_oidc_id()
+ assume_account_policy = SERVICE_ROLE_ASSUME_POLICY_TEMPLATE \
+ .format(self.account_id, oidc_id, self.push_namespace,
+ PUSH_ENDPOINT_POD_NAME)
+ permissions_policy = SERVICE_ROLE_PERMISSIONS_POLICY_TEMPLATE \
+ .format('", "'.join(queue_arns))
+ role_arn: str
+ try:
+ role = self.iam_client.create_role(RoleName=service_role_name,
+ AssumeRolePolicyDocument=assume_account_policy)
+ self.iam_client.put_role_policy(RoleName=service_role_name,
+ PolicyName=self.sa_role_policy_name,
+ PolicyDocument=permissions_policy)
+ role_arn = role["Role"]["Arn"]
+ except self.iam_client.exceptions.EntityAlreadyExistsException:
+ role = self.iam_client.get_role(RoleName=service_role_name)
+ role_arn = role["Role"]["Arn"]
+ return role_arn
+
+ def create_k8s_file(self, queue_urls: typing.List[str],
+ env_file: typing.TextIO, role_arn: str):
+ queue_url_templates = []
+ for i, url in enumerate(queue_urls):
+ queue_url_templates.append(QUEUE_URL_TEMPLATE.format(i, url))
+ env_file.write(f"export PUSH_ENDPOINT_QUEUE_{i}={url}\n")
+ env_file.write(f"export PUSH_ENDPOINT_URL_{i}={self.base_url}{self.api_path}/{i}\n")
+ env_file.write(f"export PUSH_ENDPOINT_NUM_QUEUES={self.num_queues}\n")
+ queue_url_template = "".join(queue_url_templates)[1:]
+
+ hmac_secret = sp.check_output(["openssl", "rand", "-hex", "15"]).decode("utf-8").strip()
+
+ notification_command = f'kubectl get deployment -n {self.core_namespace} os-notification --output json | jq -r ".spec.template.spec.containers[0].image"'
+ notification_result = sp.run(notification_command, shell=True, capture_output=True, encoding="utf-8")
+ if notification_result.stderr:
+ print("Notification command had standard error:", notification_result.stderr)
+
+ notification_image = notification_result.stdout.strip()
+ image_parts = notification_image.split(":")
+ print(image_parts, notification_image)
+ if len(image_parts) != 2:
+ raise RuntimeError("Unexpected result from Kubectl command.")
+
+ endpoint_image_tag = f"push_endpoint.{image_parts[1].split('.', 1)[1]}"
+ endpoint_image = f"{image_parts[0]}:{endpoint_image_tag[:120]}"
+ env_file.write(f"export HMAC_SECRET={hmac_secret}\n")
+
+ allowed_origins = [
+ ALLOWED_ORIGINS_TEMPLATE.format("http://localhost"),
+ ALLOWED_ORIGINS_TEMPLATE.format(f"https://{self.domain_name}")
+ ]
+ print(allowed_origins, endpoint_image)
+
+ values_contents = PUSH_ENDPOINT_K8S_TEMPLATE \
+ .format(core_namespace=self.core_namespace,
+ ingress_namespace=self.ingress_namespace,
+ push_namespace=self.push_namespace,
+ ingress_gateway_name=self.ingress_gateway,
+ osdu_gateway_name=self.osdu_gateway,
+ name=PUSH_ENDPOINT_POD_NAME,
+ hmac_secret=hmac_secret,
+ aws_region=self.region,
+ num_queues=self.num_queues,
+ image=endpoint_image,
+ queue_urls=queue_url_template,
+ api_path=self.api_path,
+ allowed_origins="".join(allowed_origins)[1:],
+ role_arn=role_arn)
+
+ with open(self.kube_file, "w") as file:
+ file.write(values_contents)
+
+ async def ensure_queue_created(self, queue_num: int,
+ received_queues: asyncio.Queue):
+ queue_name = self.get_sqs_queue_name(queue_num)
+ try:
+ queue = self.sqs_client.create_queue(QueueName=queue_name)
+ except self.sqs_client.exceptions.QueueDeletedRecently:
+ time.sleep(62)
+ await self.ensure_queue_created(queue_num, received_queues)
+ return
+ except self.sqs_client.exceptions.QueueNameExists:
+ queue = self.sqs_client.get_queue_url(QueueName=queue_name)
+ await received_queues.put(queue["QueueUrl"])
+
+ def get_sqs_queue_arn(self, queue_url: str):
+ QUEUE_ARN_ATTR = "QueueArn"
+ response = self.sqs_client.get_queue_attributes(
+ QueueUrl=queue_url,
+ AttributeNames=[QUEUE_ARN_ATTR]
+ )
+ return response["Attributes"][QUEUE_ARN_ATTR]
+
+ def disable_jwt_auth_on_path(self):
+ output = sp.check_output(["kubectl", "get", "authorizationpolicy", "-n",
+ self.ingress_namespace, "ingress-jwt-required",
+ "-o", "json"])
+ current_auth_policy = json.loads(output)
+ for rule in current_auth_policy["spec"]["rules"]:
+ for outer_op in rule["to"]:
+ not_paths = outer_op["operation"]["notPaths"]
+ if self.wildcard_path not in not_paths:
+ not_paths.append(self.wildcard_path)
+ new_auth_policy = json.dumps(current_auth_policy)
+ sp.check_output(["kubectl", "apply", "-f", "-"], input=new_auth_policy,
+ encoding='utf-8')
+
+ async def create_resources(self):
+ received_queues = asyncio.Queue()
+ tasks = []
+ for i in range(self.num_queues):
+ task = asyncio.create_task(
+ self.ensure_queue_created(i, received_queues)
+ )
+ tasks.append(task)
+
+ await asyncio.gather(*tasks)
+ print("Should have created the queues.")
+
+ queue_arns = []
+ queue_urls = []
+ while not received_queues.empty():
+ queue_url = await received_queues.get()
+ queue_arns.append(self.get_sqs_queue_arn(queue_url))
+ queue_urls.append(queue_url)
+
+ role_arn = self.create_policy_document(queue_arns)
+ print("Should have created the IAM Role and policy. ARN:", role_arn)
+
+ with open(self.env_file, "w") as file:
+ self.create_k8s_file(queue_urls, file, role_arn)
+
+ print("Now creating the Kubernetes resources")
+ sp.check_output(["kubectl", "apply", "-f", self.kube_file])
+
+ print("Now changing the JWT Authorization Policy")
+ self.disable_jwt_auth_on_path()
+
+ async def ensure_queue_deleted(self, queue_num: int):
+ queue_name = self.get_sqs_queue_name(queue_num)
+ try:
+ queue_url = self.sqs_client.get_queue_url(QueueName=queue_name)
+ self.sqs_client.delete_queue(QueueUrl=queue_url["QueueUrl"])
+ except self.sqs_client.exceptions.QueueDoesNotExist:
+ print("Queue", queue_name, "not found, so could not delete!")
+
+ def re_enable_jwt_auth_on_path(self):
+ output = sp.check_output(["kubectl", "get", "authorizationpolicy", "-n",
+ self.ingress_namespace, "ingress-jwt-required",
+ "-o", "json"])
+ current_auth_policy = json.loads(output)
+ for rule in current_auth_policy["spec"]["rules"]:
+ for outer_op in rule["to"]:
+ not_paths = outer_op["operation"]["notPaths"]
+ try:
+ not_paths.remove(self.wildcard_path)
+ except ValueError:
+ print(f"Path {self.wildcard_path} was already not exempted from the disable JWT authentication.")
+
+ new_auth_policy = json.dumps(current_auth_policy)
+ sp.check_output(["kubectl", "apply", "-f", "-"], input=new_auth_policy,
+ encoding='utf-8')
+
+ async def delete_resources(self):
+ print("Now changing the JWT Authorization Policy")
+ self.re_enable_jwt_auth_on_path()
+
+ print("Deleting the kubernetes resources.")
+ try:
+ sp.check_output(["kubectl", "delete", "-f", self.kube_file])
+ except Exception as e:
+ print("Failed to delete the kubernetes resources:", e)
+
+ print("Deleting the role policy.")
+ try:
+ self.iam_client.delete_role_policy(RoleName=self.get_service_role(),
+ PolicyName=self.sa_role_policy_name)
+ self.iam_client.delete_role(RoleName=self.get_service_role())
+ except self.iam_client.exceptions.NoSuchEntityException:
+ print("Could not delete role", self.get_service_role(),
+ "because it does not exist!")
+
+ print("Deleting the SQS queues.")
+ tasks = []
+ for i in range(self.num_queues):
+ task = asyncio.create_task(self.ensure_queue_deleted(i))
+ tasks.append(task)
+
+ await asyncio.gather(*tasks)
+
+
+async def main():
+ argparser = ap.ArgumentParser(description="Handles creation/deletion of the"
+ " AWS resources required for the integration te"
+ "sts and creates an environment variable file f"
+ "or the integration tests to use.")
+ argparser.add_argument('--region', type=str,
+ default=os.getenv('AWS_REGION'),
+ help="The AWS region to deploy the cloud resources to.")
+ argparser.add_argument('--num_queues', type=int, default=2,
+ help="The number of queues to create.")
+ argparser.add_argument('--osdu_instance_name', type=str,
+ default=os.getenv("OSDU_INSTANCE_NAME"),
+ help="The OSDU instance name.")
+ argparser.add_argument('--eks_cluster_name', type=str,
+ default=os.getenv("EKS_CLUSTER_NAME"))
+ argparser.add_argument('--base_url', type=str,
+ default=os.getenv("AWS_BASE_URL"))
+ argparser.add_argument('--domain_name', type=str,
+ default=os.getenv("DOMAIN"))
+ argparser.add_argument('--endpoint_suffix', type=str,
+ default=NOTIFICATION_SUFFIX)
+ argparser.add_argument('--kube_file', type=str, default=None)
+ argparser.add_argument('--api_path', type=str,
+ default=PUSH_ENDPOINT_PATH,
+ help="The push endpoint path relative from the ingress gateway.")
+ argparser.add_argument('--environment_file', type=str,
+ default="./environment_file.sh",
+ help="The file to create the environment variables in.")
+ argparser.add_argument('--create', action='store_true', help="Creates the "
+ "resources required for the integration tests.")
+ argparser.add_argument('--delete', action='store_true', help="Deletes the "
+ "resources created for the integration tests.")
+
+ args = argparser.parse_args()
+ if args.create and args.delete:
+ raise ValueError("Cannot create and delete at the same time.")
+
+ if not args.create and not args.delete:
+ raise ValueError("Must specify either create or delete.")
+
+ osdu_instance = args.osdu_instance_name
+ eks_cluster = args.eks_cluster_name
+ num_queues = args.num_queues
+ region = args.region
+ env_file = args.environment_file
+ base_url = args.base_url
+ domain_name = args.domain_name
+ endpoint_suffix = args.endpoint_suffix
+ kube_file = args.kube_file
+ api_path = args.api_path
+ common_params = CommonParameters(osdu_instance, eks_cluster, region,
+ num_queues, env_file, base_url, domain_name,
+ endpoint_suffix, kube_file, api_path)
+
+ if args.create:
+ await common_params.create_resources()
+
+ if args.delete:
+ await common_params.delete_resources()
+
+
+if __name__ == '__main__':
+ asyncio.run(main())
+
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/Application.java b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/Application.java
new file mode 100644
index 0000000000000000000000000000000000000000..af98f486137d175490ae38c39bd7d73fdb0615c8
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/Application.java
@@ -0,0 +1,30 @@
+/* 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.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 {
+
+ public static void main(String[] args) {
+ SpringApplication.run(Application.class, args);
+ }
+}
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/HmacVerification.java b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/HmacVerification.java
new file mode 100644
index 0000000000000000000000000000000000000000..3a730f523414cbf48d6c3d364db53bdef6693c57
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/HmacVerification.java
@@ -0,0 +1,71 @@
+/* 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.example;
+
+import org.opengroup.osdu.core.common.cryptographic.SignatureService;
+import org.opengroup.osdu.core.common.cryptographic.SignatureServiceException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.web.context.annotation.RequestScope;
+
+import jakarta.servlet.http.HttpServletRequest;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Map;
+
+@Component("hmacVerification")
+@RequestScope
+public class HmacVerification {
+ @Autowired
+ private HttpServletRequest request;
+
+ private final SignatureService service = new SignatureService();
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(HmacVerification.class);
+
+ public boolean verifyHmac() {
+ boolean foundError = false;
+ LOGGER.info("Inside verifyHmac");
+ LOGGER.error("Request: {}", request);
+ try {
+ Map<String, String> params = QueryStringUtils.parseQueryString(request.getQueryString());
+ LOGGER.info(request.getQueryString());
+ String tokenToVerify = params.get("hmac");
+ if (tokenToVerify == null) throw new SignatureServiceException("Hmac verification not attached to request!");
+ tokenToVerify = URLDecoder.decode(tokenToVerify, StandardCharsets.UTF_8);
+ service.verifyHmacSignature(tokenToVerify, OSDUSecretString.getSecret());
+ } catch (SignatureServiceException except) {
+ LOGGER.error("Failed to verify request.", except);
+ foundError = true;
+ }
+ return !foundError;
+ }
+
+ private static final String URL = "https://my.url/com";
+ private static final String NONCE = "FEDCBA9876543210";
+ public static void main(String[] args) throws SignatureServiceException {
+ SignatureService service = new SignatureService();
+ String expireTime = Long.toString(System.currentTimeMillis() + 3600000);
+ String data = String.format("{\"expireMillisecond\": \"%s\",\"hashMechanism\": \"hmacSHA256\",\"endpointUrl\": \"%s\",\"nonce\": \"%s\"}", expireTime, URL, NONCE);
+ String encodedData = Base64.getEncoder().encodeToString(data.getBytes(StandardCharsets.UTF_8));
+ String signature = encodedData + "." + service.getSignedSignature(URL, OSDUSecretString.getSecret(), expireTime, NONCE);
+ System.out.println(signature);
+ }
+}
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/OSDUSecretString.java b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/OSDUSecretString.java
new file mode 100644
index 0000000000000000000000000000000000000000..cef67c18a5e6160c7bd5204e4151b0f212a80daf
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/OSDUSecretString.java
@@ -0,0 +1,32 @@
+/* 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.example;
+
+public class OSDUSecretString {
+ private static String secretValue = null;
+ //private static String secretValue = "0123456789abcdef";
+ private static final String SECRET_ENV_NAME = "HMAC_SECRET";
+
+ public static String getSecret() {
+ if (secretValue == null) {
+ secretValue = System.getenv(SECRET_ENV_NAME);
+ if (secretValue == null) throw new NullPointerException("Must specify the HMAC Secret via the environment variable: " + SECRET_ENV_NAME);
+ }
+ System.out.println("Got secret...");
+ return secretValue;
+ }
+
+}
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/PushEndpointController.java b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/PushEndpointController.java
new file mode 100644
index 0000000000000000000000000000000000000000..12020776d0e873b60282ea946a514f19d8615ad2
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/PushEndpointController.java
@@ -0,0 +1,126 @@
+/* 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.example;
+
+import com.amazonaws.services.sqs.AmazonSQS;
+import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
+import com.amazonaws.services.sqs.model.SendMessageResult;
+import com.google.common.hash.Hashing;
+import org.apache.http.HttpStatus;
+import org.opengroup.osdu.core.common.model.http.AppException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.annotation.RequestScope;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.servlet.HandlerMapping;
+
+import jakarta.servlet.http.HttpServletRequest;
+import javax.validation.constraints.NotBlank;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Map;
+import java.util.Scanner;
+
+@RestController
+@RequestMapping("/")
+@RequestScope
+@Validated
+public class PushEndpointController {
+ private final AmazonSQS sqsClient;
+ private final int maxQueueNum;
+ private static final Logger LOGGER = LoggerFactory.getLogger(PushEndpointController.class);
+
+ public PushEndpointController() {
+ sqsClient = AmazonSQSClientBuilder.defaultClient();
+ maxQueueNum = Integer.parseInt(System.getenv("MAX_QUEUE_NUM"));
+ }
+
+ private static String verifySubscriptionForChallenge(String crc) {
+ String response = OSDUSecretString.getSecret() + crc;
+ response = Hashing.sha256()
+ .hashString(response, StandardCharsets.UTF_8)
+ .toString();
+ response = Base64.getEncoder().encodeToString(response.getBytes());
+ return response;
+ }
+
+ @GetMapping("/{count}")
+ @PreAuthorize("@hmacVerification.verifyHmac()")
+ public ResponseEntity<VerifySubscriptionResponse> verifySubscription(@RequestParam("crc") @NotBlank String crc, HttpServletRequest request) {
+ getQueueNumberFromRequest(request);
+ String response = verifySubscriptionForChallenge(crc);
+ VerifySubscriptionResponse returnValue = new VerifySubscriptionResponse(response);
+ LOGGER.info(returnValue.toString());
+ return ResponseEntity.ok(returnValue);
+ }
+
+ private String readMessage(HttpServletRequest request) throws IOException {
+ Scanner scan = new Scanner(request.getInputStream());
+ StringBuilder builder = new StringBuilder();
+ while (scan.hasNextLine()) {
+ builder.append(scan.nextLine());
+ }
+ return builder.toString();
+ }
+
+ private void throwAppException(String message) {
+ throw new AppException(HttpStatus.SC_BAD_REQUEST, "Bad Request", message);
+ }
+
+ private int getQueueNumberFromRequest(HttpServletRequest request) {
+ Object uriTemplateObject = request.getAttribute(HandlerMapping.URI_TEMPLATE_VARIABLES_ATTRIBUTE);
+ if (uriTemplateObject == null || !(uriTemplateObject instanceof Map)) {
+ throwAppException("Invalid request path!");
+ }
+ Map<String, String> uriTempalteVariables = (Map<String, String>) request.getAttribute(HandlerMapping.URI_TEMPLATE_VARIABLES_ATTRIBUTE);
+ String queueNumber = uriTempalteVariables.get("count");
+ int returnValue = Integer.parseInt(queueNumber);
+ if (returnValue >= maxQueueNum) {
+ throwAppException("Invalid queue number!");
+ }
+ return returnValue;
+ }
+
+ private String getQueueURL(HttpServletRequest request) {
+ String queueURL = System.getenv("QUEUE_NUM_" + getQueueNumberFromRequest(request));
+ if (queueURL == null) {
+ throwAppException("Queue not found!");
+ }
+ return queueURL;
+ }
+
+ @PostMapping("/{count}")
+ @PreAuthorize("@hmacVerification.verifyHmac()")
+ public ResponseEntity<SendMessageResult> pushNotificationToQueue(HttpServletRequest request) throws IOException {
+ String queueURL = getQueueURL(request);
+ String message = readMessage(request);
+ SendMessageResult returnValue = sqsClient.sendMessage(queueURL, message);
+ LOGGER.info(returnValue.toString());
+ return ResponseEntity.ok(returnValue);
+ }
+}
+
+record VerifySubscriptionResponse(String responseHash) {
+
+}
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/QueryStringUtils.java b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/QueryStringUtils.java
new file mode 100644
index 0000000000000000000000000000000000000000..d79e70e4f921177a080f68999d61345db83731c7
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/QueryStringUtils.java
@@ -0,0 +1,32 @@
+/* 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.example;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class QueryStringUtils {
+
+ public static Map<String, String> parseQueryString(String queryString) {
+ HashMap<String, String> retValue = new HashMap<>();
+ for (String pair : queryString.split("&")) {
+ String[] keyValue = pair.split("=", 2);
+ if (keyValue.length == 2) retValue.put(keyValue[0], keyValue[1]);
+ }
+ return retValue;
+ }
+
+}
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/SecurityConfig.java b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/SecurityConfig.java
new file mode 100644
index 0000000000000000000000000000000000000000..093ffd18ab8b6fa3fa6dde7f4dc0da626fba8c50
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/java/org/example/SecurityConfig.java
@@ -0,0 +1,34 @@
+/* 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.example;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.method.configuration.EnableGlobalMethodSecurity;
+import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.web.SecurityFilterChain;
+
+@EnableWebSecurity
+@EnableGlobalMethodSecurity(prePostEnabled = true)
+@Configuration
+public class SecurityConfig {
+
+ @Bean
+ public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
+ return http.httpBasic().disable().csrf().disable().build();
+ }
+}
diff --git a/testing/notification-test-aws/build-aws/push-endpoint/src/main/resources/application.properties b/testing/notification-test-aws/build-aws/push-endpoint/src/main/resources/application.properties
new file mode 100644
index 0000000000000000000000000000000000000000..30b07cc6788082a9b99213bc91fd967433dc1335
--- /dev/null
+++ b/testing/notification-test-aws/build-aws/push-endpoint/src/main/resources/application.properties
@@ -0,0 +1,9 @@
+LOG_PREFIX=push_endpoint
+logging.level.org.springframework.web=${LOG_LEVEL:INFO}
+server.servlet.contextPath=/api/push-endpoint
+server.port=${APPLICATION_PORT:8080}
+
+spring.application.name=push-endpoint
+management.endpoint.health.probes.enabled=true
+management.health.livenessState.enabled=true
+management.health.readinessState.enabled=true
diff --git a/testing/notification-test-aws/build-aws/run-tests.sh b/testing/notification-test-aws/build-aws/run-tests.sh
index afd1e9e3a44432dacecbd08cd37d703cac09c99c..5a181f150a5ccac46b6de6286638f4d7ecee6e93 100755
--- a/testing/notification-test-aws/build-aws/run-tests.sh
+++ b/testing/notification-test-aws/build-aws/run-tests.sh
@@ -30,6 +30,11 @@ echo "$SCRIPT_SOURCE_DIR"
# The following variables are automatically populated from the environment during integration testing
# see os-deploy-aws/build-aws/integration-test-env-variables.py for an updated list
+PUSH_ENDPOINT_PATH=$SCRIPT_SOURCE_DIR/push-endpoint
+python3 $PUSH_ENDPOINT_PATH/setup_teardown_endpoint.py --create --environment_file $PUSH_ENDPOINT_PATH/environment_file.sh --kube_file $PUSH_ENDPOINT_PATH/push-endpoint-manifest.yaml
+
+trap "python3 $PUSH_ENDPOINT_PATH/setup_teardown_endpoint.py --delete --environment_file $PUSH_ENDPOINT_PATH/environment_file.sh --kube_file $PUSH_ENDPOINT_PATH/push-endpoint-manifest.yaml" EXIT
+source $PUSH_ENDPOINT_PATH/environment_file.sh
export AWS_COGNITO_AUTH_FLOW=USER_PASSWORD_AUTH
export AWS_COGNITO_AUTH_PARAMS_PASSWORD=$ADMIN_PASSWORD
@@ -37,13 +42,16 @@ export AWS_COGNITO_AUTH_PARAMS_USER=$ADMIN_USER
export AWS_COGNITO_AUTH_PARAMS_USER_NO_ACCESS=$USER_NO_ACCESS
export ENVIRONMENT=DEV
export NOTIFICATION_REGISTER_BASE_URL=$NOTIFICATION_REGISTER_BASE_URL
-export HMAC_SECRET=02030405060708090A0B0C0D0E0F
-export REGISTER_CUSTOM_PUSH_PATH=/api/register/v1/awstest/aws/challenge/1
+export REGISTER_CUSTOM_PUSH_PATH=/api/push-endpoint/0
+export STORAGE_HOST=$STORAGE_URL
+export FILE_URL=$FILE_URL
export REGISTER_CUSTOM_PUSH_URL_HMAC=$NOTIFICATION_REGISTER_BASE_URL$REGISTER_CUSTOM_PUSH_PATH
export NOTIFICATION_BASE_URL=$NOTIFICATION_BASE_URL
-
+export OSDU_TENANT=int-test-notification
+export TENANT_NAME=int-test-notification
#### RUN INTEGRATION TEST #########################################################################
+JAVA_HOME=$JAVA17_HOME
mvn test -f "$SCRIPT_SOURCE_DIR"/../pom.xml
TEST_EXIT_CODE=$?
diff --git a/testing/notification-test-aws/pom.xml b/testing/notification-test-aws/pom.xml
index 15f79aaacb288d9a119108d8ece41b37da167702..56fa1b277a2ddbac69d53610801572d9fd34fbb2 100644
--- a/testing/notification-test-aws/pom.xml
+++ b/testing/notification-test-aws/pom.xml
@@ -51,6 +51,16 @@
<artifactId>aws-java-sdk-cognitoidp</artifactId>
<version>1.11.676</version>
</dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-ssm</artifactId>
+ <version>1.11.339</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-sqs</artifactId>
+ <version>1.11.339</version>
+ </dependency>
<dependency>
<groupId>org.opengroup.osdu.notification</groupId>
<artifactId>notification-test-core</artifactId>
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/PubsubEndpointHMACDescriptor.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/PubsubEndpointHMACDescriptor.java
index 9f439169743714ddba75eebe46620b3b6ad052cd..1417d968e90ff4c0577eb2f65dafa1a14171db42 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/PubsubEndpointHMACDescriptor.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/PubsubEndpointHMACDescriptor.java
@@ -1,15 +1,17 @@
-// Copyright © 2020 Amazon Web Services
-// 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.
+/* 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.api;
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/TestPubsubEndpointHMAC.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/TestPubsubEndpointHMAC.java
index 5627cbc10ce458d23c6a11096f1dc39318835dae..2624baffad719e8275ec1957bd7c9d8776903a17 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/TestPubsubEndpointHMAC.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/api/TestPubsubEndpointHMAC.java
@@ -1,16 +1,18 @@
-// Copyright © 2020 Amazon Web Services
-//
-// 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.
+/* 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.api;
import static org.junit.Assert.assertEquals;
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/FileTestUtils.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/FileTestUtils.java
new file mode 100644
index 0000000000000000000000000000000000000000..228735972a92a98066445151e488299e5e8ad5a5
--- /dev/null
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/FileTestUtils.java
@@ -0,0 +1,321 @@
+/* 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 static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.amazonaws.services.sqs.model.Message;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.sun.jersey.api.client.Client;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import org.apache.commons.lang3.tuple.Pair;
+import org.opengroup.osdu.core.common.model.entitlements.Acl;
+import org.opengroup.osdu.core.common.model.file.LocationResponse;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.legal.Legal;
+import org.opengroup.osdu.notification.util.AwsTestUtils;
+
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+import javax.ws.rs.core.MediaType;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.security.SecureRandom;
+import java.security.cert.X509Certificate;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+public class FileTestUtils {
+
+ private final DpsHeaders dpsHeaders;
+ private final Client restClient;
+ private final String baseUrl;
+ private final static String DATASET_KIND = "osdu:wks:dataset--File.Generic:1.0.0";
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ private static class NullTrustManager implements X509TrustManager {
+ @Override
+ public X509Certificate[] getAcceptedIssuers() {
+ return null;
+ }
+
+ @Override
+ public void checkClientTrusted(X509Certificate[] certs, String authType) {
+ }
+
+ @Override
+ public void checkServerTrusted(X509Certificate[] certs, String authType) {
+ }
+ }
+
+ private record FileMetadataResponse(String id) {
+ }
+
+ public FileTestUtils(DpsHeaders headers) {
+ TrustManager[] trustAllCerts = new TrustManager[]{new NullTrustManager()};
+
+ try {
+ SSLContext sc = SSLContext.getInstance("TLS");
+ sc.init(null, trustAllCerts, new SecureRandom());
+ HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
+ } catch (Exception e) {
+ }
+ restClient = Client.create();
+ dpsHeaders = headers;
+ baseUrl = System.getProperty("FILE_URL", System.getenv("FILE_URL"));
+ assertNotEquals(null, baseUrl, "Must set the FILE_URL environment variable!");
+ }
+
+ /**
+ * Returns the expected message values for the given the legal tag name to use.
+ * @param legalTagName The legal tag name.
+ * @return The expected message values. Used to compare with the actual message values.
+ */
+ public String uploadFile(String legalTagName) throws JsonProcessingException {
+ LocationResponse locationResponse = getFileLocation();
+ String signedUrl = locationResponse.getLocation().get("SignedURL");
+ assertNotEquals(null, signedUrl);
+ String fileSource = locationResponse.getLocation().get("FileSource");
+ uploadTestFile(signedUrl);
+ return setFileMetadata(fileSource, legalTagName);
+ }
+
+ private String getFileMetadataURL() {
+ try {
+ URL mergedURL = new URL(baseUrl + "/files/metadata");
+ return mergedURL.toString();
+ } catch (MalformedURLException e) {
+ fail("Failed to get file metadata URL");
+ }
+ return null;
+ }
+
+ private String getFileUploadURL() {
+ try {
+ URL mergedURL = new URL(baseUrl + "/files/uploadURL");
+ return mergedURL.toString();
+ } catch (MalformedURLException e) {
+ fail("Failed to get file upload URL");
+ }
+ return null;
+ }
+
+ 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);
+ }
+
+ private LocationResponse getFileLocation() throws JsonProcessingException {
+ ClientResponse response = sendRequest(getFileUploadURL(), "GET", true, null, null);
+ assertEquals(200, response.getStatus());
+ String responseData = response.getEntity(String.class);
+ return objectMapper.readValue(responseData, LocationResponse.class);
+ }
+
+ private void uploadTestFile(String url) {
+ ClientResponse response = sendRequest(url, "PUT", false, "Test file upload.", MediaType.MULTIPART_FORM_DATA_TYPE);
+ assertEquals(200, response.getStatus());
+ }
+
+ private JsonObject getFileMetadataData(String fileSource) {
+ JsonObject dataObject = new JsonObject();
+ dataObject.addProperty("Endian", "BIG");
+ dataObject.addProperty("Description", "Test file");
+ dataObject.addProperty("TotalSize", "1048576");
+ JsonObject fileSourceInfo = new JsonObject();
+ fileSourceInfo.addProperty("FileSource", fileSource);
+ fileSourceInfo.addProperty("Name", "1234");
+ fileSourceInfo.addProperty("PreloadFilePath", "gs://osdu-cicd-epam-persistent-area/c3af38c1-654d-47a0-a3e6-9e94c32add84/b62b104f843142f49ee6d747e6bdd49d");
+ fileSourceInfo.addProperty("PreloadFileCreateUser", "user1");
+ fileSourceInfo.addProperty("PreloadFileModifyDate", "mar 11");
+ fileSourceInfo.addProperty("PreloadFileModifyUser", "mar 11");
+ JsonObject datasetProperties = new JsonObject();
+ datasetProperties.add("FileSourceInfo", fileSourceInfo);
+ dataObject.add("DatasetProperties", datasetProperties);
+ return dataObject;
+ }
+
+ private JsonArray getJsonArrayFromStringArray(Iterable<String> input) {
+ JsonArray jsonArray = new JsonArray();
+ for (String str : input) {
+ jsonArray.add(str);
+ }
+ return jsonArray;
+ }
+
+ private JsonObject getFileMetadataBase(String legalTagName) {
+ Pair<Acl, Legal> aclLegalPair = RecordUtils.getAclAndLegal(RecordUtils.getDefaultAcls(true), legalTagName, "US");
+ JsonObject jsonAcl = new JsonObject();
+ Acl acl = aclLegalPair.getLeft();
+ jsonAcl.add("viewers", getJsonArrayFromStringArray(Arrays.asList(acl.viewers)));
+ jsonAcl.add("owners", getJsonArrayFromStringArray(Arrays.asList(acl.owners)));
+
+ Legal legal = aclLegalPair.getRight();
+ JsonObject jsonLegal = new JsonObject();
+ jsonLegal.add("legaltags", getJsonArrayFromStringArray(legal.getLegaltags()));
+ jsonLegal.add("otherRelevantDataCountries", getJsonArrayFromStringArray(legal.getOtherRelevantDataCountries()));
+ jsonLegal.addProperty("status", "compliant");
+
+ JsonObject fileMetadata = new JsonObject();
+ fileMetadata.add("acl", jsonAcl);
+ fileMetadata.add("legal", jsonLegal);
+ return fileMetadata;
+ }
+
+ private JsonObject getFileMetadata(String fileSource, String legalTagName) {
+ JsonObject fileMetadata = getFileMetadataBase(legalTagName);
+ fileMetadata.addProperty("createUser", "osdu-community-sa-airflow@nice-etching-277309.iam.gserviceaccount.com");
+ fileMetadata.addProperty("createTime", "2021-02-22T18:50:47.498Z");
+ fileMetadata.addProperty("modifyUser", "osdu-community-sa-airflow@nice-etching-277309.iam.gserviceaccount.com");
+ fileMetadata.addProperty("modifyTime", "2021-02-22T21:13:10.587Z");
+ fileMetadata.addProperty("kind", DATASET_KIND);
+
+ JsonObject dataObject = getFileMetadataData(fileSource);
+ fileMetadata.add("data", dataObject);
+
+ return fileMetadata;
+ }
+
+ private String setFileMetadata(String fileSource, String legalTagName) throws JsonProcessingException {
+ JsonObject fileMetadata = getFileMetadata(fileSource, legalTagName);
+
+ ClientResponse response = sendRequest(getFileMetadataURL(), "POST", true, fileMetadata.toString(), null);
+ assertEquals(201, response.getStatus());
+ String responseData = response.getEntity(String.class);
+ return objectMapper.readValue(responseData, FileMetadataResponse.class).id();
+ }
+
+ private void compareVersionIdWithExpected(String fileId, Object versionId, Instant started, Instant ended) {
+ assertNotEquals(null, versionId);
+ assertTrue(versionId instanceof String);
+ String versionString = versionId.toString();
+ assertTrue(versionString.startsWith(fileId));
+ String versionIdString = versionString.substring(fileId.length() + 1);
+ long versionIdMicros = Long.parseLong(versionIdString);
+ final long MICROS_PER_SECOND = 1000000;
+ final long NANOS_PER_MICRO = 1000;
+ long versionIdNanos = (versionIdMicros % MICROS_PER_SECOND) * NANOS_PER_MICRO;
+ long versionIdSeconds = versionIdMicros / MICROS_PER_SECOND;
+ Instant versionIdInstant = Instant.ofEpochSecond(versionIdSeconds, versionIdNanos);
+ assertTrue(versionIdInstant.isAfter(started));
+ assertTrue(versionIdInstant.isBefore(ended));
+ }
+
+ private void compareTimestamp(Object timestampObj, Instant started, Instant ended) {
+ assertNotEquals(null, timestampObj);
+ assertTrue(timestampObj instanceof Number);
+ Instant instantTs = Instant.ofEpochMilli(((Number) timestampObj).longValue());
+ assertTrue(instantTs.isAfter(started));
+ assertTrue(instantTs.isBefore(ended));
+ }
+
+ private void checkDatasetDetails(Map<String, Object> properties, String fileId, Instant started, Instant ended) {
+ assertEquals(fileId, properties.get("datasetId"));
+ compareVersionIdWithExpected(fileId, properties.get("datasetVersionId"), started, ended);
+ assertEquals("FILE", properties.get("datasetType").toString());
+ Object recordCount = properties.get("recordCount");
+ assertTrue(recordCount instanceof Number);
+ assertEquals(1, recordCount);
+ compareTimestamp(properties.get("timestamp"), started, ended);
+ }
+
+ private void checkInProgressStatus(Map<String, Object> properties, Instant started, Instant ended) {
+ Object errorCode = properties.get("errorCode");
+ assertNotEquals(null, errorCode);
+ assertTrue(errorCode instanceof Number);
+ assertEquals(0, errorCode);
+ assertEquals(dpsHeaders.getUserId(), properties.get("userEmail"));
+ compareTimestamp(properties.get("timestamp"), started, ended);
+ Object message = properties.get("message");
+ assertNotEquals(null, message);
+ assertTrue(message instanceof String);
+ }
+
+ private void checkSuccessStatus(Map<String, Object> properties, String fileId, Instant started, Instant ended) {
+ checkInProgressStatus(properties, started, ended);
+ assertEquals(fileId, properties.get("recordId"));
+ compareVersionIdWithExpected(fileId, properties.get("recordIdVersion"), started, ended);
+ }
+
+ public void compareMessagesWithExpected(String fileId, List<Message> receivedMessages, Instant started) throws Exception {
+ Instant ended = Instant.now();
+ assertEquals("Expect exactly 3 messages: dataset sync start, dataset sync end, and dataset details", 3, receivedMessages.size());
+ String correlationId = null;
+ boolean syncStartFound = false;
+ boolean syncEndFound = false;
+ boolean datasetDetailsFound = false;
+ for (Message message : receivedMessages) {
+ FileMessage fileMessage = AwsTestUtils.unwrapFirstMessage(message, FileMessage[].class);
+ Map<String, Object> properties = fileMessage.properties();
+ if (correlationId == null) {
+ correlationId = properties.get("correlationId").toString();
+ } else {
+ assertEquals(correlationId, properties.get("correlationId").toString());
+ }
+
+
+ if (fileMessage.kind().equals("datasetDetails")) {
+ assertFalse(datasetDetailsFound);
+ datasetDetailsFound = true;
+ checkDatasetDetails(properties, fileId, started, ended);
+ } else if (fileMessage.kind().equals("status")) {
+ Object status = properties.get("status");
+ Object stage = properties.get("stage");
+ assertNotEquals(null, status);
+ assertNotEquals(null, stage);
+ assertTrue(status instanceof String);
+ assertTrue(stage instanceof String);
+ assertEquals("DATASET_SYNC", stage);
+ if (status.equals("IN_PROGRESS")) {
+ assertFalse(syncStartFound);
+ syncStartFound = true;
+ checkInProgressStatus(properties, started, ended);
+ } else if (status.equals("SUCCESS")) {
+ assertFalse(syncEndFound);
+ syncEndFound = true;
+ checkSuccessStatus(properties, fileId, started, ended);
+ } else {
+ fail("Unexpected status: " + status);
+ }
+ } else {
+ fail("Unexpected kind: " + fileMessage.kind());
+ }
+ }
+ }
+}
+
+record FileMessage (String kind, Map<String, Object> properties) {
+}
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/LegalTagUtils.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/LegalTagUtils.java
index 552bd5bd0c798eb3136552cfb8a374388502f070..a0d2c2584a0e9d024407f269f73cc5a7c7a67749 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/LegalTagUtils.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/LegalTagUtils.java
@@ -1,76 +1,101 @@
-// Copyright © Amazon Web Services
-//
-// 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.
-
+/* 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 org.apache.http.HttpStatus;
-
-import com.google.gson.JsonArray;
-import com.google.gson.JsonObject;
-import com.sun.jersey.api.client.ClientResponse;
+import org.opengroup.osdu.core.common.http.json.HttpResponseBodyMapper;
+import org.opengroup.osdu.core.common.legal.ILegalProvider;
+import org.opengroup.osdu.core.common.legal.LegalAPIConfig;
+import org.opengroup.osdu.core.common.legal.LegalFactory;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.legal.LegalException;
+import org.opengroup.osdu.core.common.model.legal.LegalTag;
+import org.opengroup.osdu.core.common.model.legal.Properties;
+import com.amazonaws.services.sqs.model.Message;
+import org.opengroup.osdu.notification.util.AwsTestUtils;
+import org.opengroup.osdu.notification.util.Config;
import org.opengroup.osdu.notification.util.TestUtils;
-import java.util.UUID;
+import java.sql.Date;
+import java.util.Collections;
+import java.util.List;
public class LegalTagUtils {
- public static ClientResponse create(String legalTagName, String token, boolean isTestPartition) throws Exception {
- return create("US", legalTagName, "2099-01-25", "Public Domain Data", token, isTestPartition);
- }
+ private final ILegalProvider legalService;
- protected static ClientResponse create(String countryOfOrigin, String name, String expDate, String dataType, String token, boolean isTestPartition)
- throws Exception {
- String body = getBody(countryOfOrigin, name, expDate, dataType);
- ClientResponse response = StorageTestUtils.send(getLegalUrl(), "legaltags", "POST", StorageTestUtils.getHeaders(TenantUtils.getTenantName(isTestPartition), token, UUID.randomUUID().toString(), isTestPartition), body,
- "");
+ public LegalTagUtils(DpsHeaders dpsHeaders, HttpResponseBodyMapper responseBodyMapper) {
+ LegalAPIConfig legalConfig = LegalAPIConfig.builder().rootUrl(Config.Instance().LegalServicePath).build();
+ LegalFactory legalFactory = new LegalFactory(legalConfig, responseBodyMapper);
+ legalService = legalFactory.create(dpsHeaders);
+ }
- assertEquals(HttpStatus.SC_CREATED, response.getStatus());
- Thread.sleep(100);
- return response;
+ public void addLegalTag(String name, String expDate) throws LegalException {
+ LegalTag legalTag = createLegalTag(name, expDate);
+ legalService.create(legalTag);
}
- public static ClientResponse delete(String legalTagName, String token, boolean isTestPartition) throws Exception {
- return StorageTestUtils.send(getLegalUrl(), "legaltags/" + legalTagName, "DELETE", StorageTestUtils.getHeaders(TenantUtils.getTenantName(isTestPartition), token, UUID.randomUUID().toString(), isTestPartition), "", "");
+ public void addLegalTag(String name) throws LegalException {
+ addLegalTag(name, "2099-01-25");
}
- protected static String getLegalUrl() {
- String legalUrl = System.getProperty("LEGAL_URL", System.getenv("LEGAL_URL"));
- return legalUrl;
+ public LegalMessage deleteLegalTag(String name) throws LegalException {
+ legalService.delete(name);
+ LegalMessageElement element = new LegalMessageElement(TestUtils.getOsduTenant(), name, "incompliant");
+ return new LegalMessage(Collections.singletonList(element));
}
- protected static String getBody(String countryOfOrigin, String name, String expDate, String dataType) {
+ public void ensureDeleted(String name) {
+ try {
+ deleteLegalTag(name);
+ } catch (LegalException e) {
+ System.err.println(e.toString());
+ }
+ }
- JsonArray coo = new JsonArray();
- coo.add(countryOfOrigin);
+ public void assertFirstMessagesSimilar(LegalMessage expected, List<Message> receivedMessages) throws Exception {
+ LegalMessage legalMessage = AwsTestUtils.unwrapSingleFirst(receivedMessages, LegalMessage.class);
+ assertEquals(1, legalMessage.statusChangedTags().size());
+ LegalMessageElement expectedElement = expected.statusChangedTags().get(0);
+ LegalMessageElement actualElement = legalMessage.statusChangedTags().get(0);
+ assertEquals(expectedElement.dataPartitionId(), actualElement.dataPartitionId());
+ assertEquals(expectedElement.changedTagName(), actualElement.changedTagName());
+ assertEquals(expectedElement.changedTagStatus(), actualElement.changedTagStatus());
+ }
- JsonObject properties = new JsonObject();
- properties.add("countryOfOrigin", coo);
- properties.addProperty("contractId", "A1234");
- properties.addProperty("expirationDate", expDate);
- properties.addProperty("dataType", dataType);
- properties.addProperty("originator", "MyCompany");
- properties.addProperty("securityClassification", "Public");
- properties.addProperty("exportClassification", "EAR99");
- properties.addProperty("personalData", "No Personal Data");
+ private LegalTag createLegalTag(String name, String expDate) {
+ LegalTag legalTag = new LegalTag();
+ legalTag.setName(name);
+ legalTag.setDescription(String.format("test for %s", name));
+ Properties legalProps = new Properties();
+ legalProps.setCountryOfOrigin(Collections.singletonList("US"));
+ legalProps.setContractId("A1234");
+ legalProps.setDataType("Public Domain Data");
+ legalProps.setExpirationDate(Date.valueOf(expDate));
+ legalProps.setOriginator("MyCompany");
+ legalProps.setSecurityClassification("Public");
+ legalProps.setExportClassification("EAR99");
+ legalProps.setPersonalData("No Personal Data");
+ legalTag.setProperties(legalProps);
+ return legalTag;
+ }
+}
- JsonObject tag = new JsonObject();
- tag.addProperty("name", name);
- tag.addProperty("description", "test for " + name);
- tag.add("properties", properties);
+record LegalMessage(List<LegalMessageElement> statusChangedTags) {
+}
- return tag.toString();
- }
+record LegalMessageElement(String dataPartitionId, String changedTagName, String changedTagStatus) {
}
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/RecordUtils.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/RecordUtils.java
index 18eed3f16243b77349b79b5ad74580df84227c97..d988ff977b11fb47cd41ade392dd54e23e3b61e0 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/RecordUtils.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/RecordUtils.java
@@ -1,93 +1,109 @@
-// Copyright © Amazon Web Services
-//
-// 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.
-
+/* 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 com.google.gson.JsonArray;
-import com.google.gson.JsonObject;
+import org.apache.commons.lang3.tuple.Pair;
import org.opengroup.osdu.core.common.Constants;
+import org.opengroup.osdu.core.common.model.entitlements.Acl;
+import org.opengroup.osdu.core.common.model.legal.Legal;
+import org.opengroup.osdu.core.common.model.storage.Record;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
public class RecordUtils {
private static final String domain = System.getProperty("DOMAIN", System.getenv("DOMAIN"));
- public static String createJsonRecordWithReference(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, String fromCrs, String conversionType, boolean isTestPartition) {
- JsonArray records = new JsonArray();
+ Record[] records = new Record[recordsCount];
for (int i = 0; i < recordsCount; i++) {
- JsonObject data = new JsonObject();
- data.addProperty("X", 16.00);
- data.addProperty("Y", 10.00);
- data.addProperty("Z", 0.0);
+ Map<String, Object> data = new HashMap<>();
+ data.put("X", 16.00);
+ data.put("Y", 10.00);
+ data.put("Z", 0.0);
JsonArray propertyNames = new JsonArray();
propertyNames.add("X");
propertyNames.add("Y");
propertyNames.add("Z");
- JsonObject meta = new JsonObject();
- meta.addProperty(Constants.KIND, conversionType);
- meta.addProperty(Constants.PERSISTABLE_REFERENCE, fromCrs);
- meta.add(Constants.PROPERTY_NAMES, propertyNames);
+ Map<String, Object> meta = new HashMap<>();
+ meta.put(Constants.KIND, conversionType);
+ meta.put(Constants.PERSISTABLE_REFERENCE, fromCrs);
+ meta.put(Constants.PROPERTY_NAMES, propertyNames);
- JsonArray metaBlocks = new JsonArray();
- metaBlocks.add(meta);
+ Map<String, Object>[] metaBlocks = new Map[1];
+ metaBlocks[0] = meta;
- JsonObject record = getRecordWithInputData(id + i, kind, legalTag, data, isTestPartition);
- record.add(Constants.META, metaBlocks);
+ Record record = getRecordWithInputData(id + i, kind, legalTag, data, isTestPartition);
+ record.setMeta(metaBlocks);
- records.add(record);
+ records[i] = record;
}
- return records.toString();
+ return records;
}
- private static JsonObject getRecordWithInputData(String id, String kind, String legalTag, JsonObject data, boolean isTestPartition) {
- JsonObject record = getDefaultRecord(id, kind, legalTag, isTestPartition);
- record.add("data", data);
+ private static Record getRecordWithInputData(String id, String kind, String legalTag, Map<String, Object> data, boolean isTestPartition) {
+ Record record = getDefaultRecord(id, kind, legalTag, isTestPartition);
+ record.setData(data);
return record;
}
- private static JsonObject getDefaultRecord(String id, String kind, String legalTag, boolean isTestPartition) {
- JsonArray acls = new JsonArray();
- acls.add(String.format("data.test1@%s", getAclSuffix(isTestPartition)));
- return getDefaultRecordFromAcl(id, kind, legalTag, acls);
+ public static String[] getDefaultAcls(boolean isTestPartition) {
+ return new String[]{String.format("data.test1@%s", getAclSuffix(isTestPartition))};
+ }
+
+ private static Record getDefaultRecord(String id, String kind, String legalTag, boolean isTestPartition) {
+ return getDefaultRecordFromAcl(id, kind, legalTag, getDefaultAcls(isTestPartition));
}
- private static JsonObject getDefaultRecordFromAcl(String id, String kind, String legalTag, JsonArray acls) {
- JsonObject acl = new JsonObject();
- acl.add("viewers", acls);
- acl.add("owners", acls);
+ private static Record getDefaultRecordFromAcl(String id, String kind, String legalTag, String[] acls) {
+ Pair<Acl, Legal> pair = getAclAndLegal(acls, legalTag, "BR");
- JsonArray tags = new JsonArray();
- tags.add(legalTag);
+ Record record = new Record();
+ record.setAcl(pair.getLeft());
+ record.setLegal(pair.getRight());
+ record.setId(id);
+ record.setKind(kind);
+ return record;
+ }
- JsonArray ordcJson = new JsonArray();
- ordcJson.add("BR");
+ public static Pair<Acl, Legal> getAclAndLegal(String[] acls, String legalTags, String... countries) {
+ Acl acl = new Acl();
+ acl.setOwners(acls);
+ acl.setViewers(acls);
- JsonObject legal = new JsonObject();
- legal.add("legaltags", tags);
- legal.add("otherRelevantDataCountries", ordcJson);
+ Set<String> ordc = new HashSet<>();
+ Collections.addAll(ordc, countries);
+ Set<String> tags = new HashSet<>();
+ tags.add(legalTags);
- JsonObject record = new JsonObject();
- record.addProperty("id", id);
- record.addProperty("kind", kind);
- record.add("acl", acl);
- record.add("legal", legal);
- return record;
+ Legal legal = new Legal();
+ legal.setLegaltags(tags);
+ legal.setOtherRelevantDataCountries(ordc);
+
+ return Pair.of(acl, legal);
}
public static String getAclSuffix(boolean isTestPartition) {
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/StorageTestUtils.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/StorageTestUtils.java
index a16554de66799d15ac3240c268a351be1fa83478..ca652f413c256efc0a6747ce68534eced2101346 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/StorageTestUtils.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/StorageTestUtils.java
@@ -1,135 +1,70 @@
-// Copyright © Amazon Web Services
-//
-// 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.
-
+/* 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 com.sun.jersey.api.client.Client;
-import com.sun.jersey.api.client.ClientResponse;
-import com.sun.jersey.api.client.WebResource;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.amazonaws.services.sqs.model.Message;
+import org.opengroup.osdu.core.common.http.json.HttpResponseBodyMapper;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.storage.Record;
+import org.opengroup.osdu.core.common.model.storage.StorageException;
+import org.opengroup.osdu.core.common.storage.IStorageService;
+import org.opengroup.osdu.core.common.storage.StorageAPIConfig;
+import org.opengroup.osdu.core.common.storage.StorageFactory;
+import org.opengroup.osdu.notification.util.AwsTestUtils;
+import org.opengroup.osdu.notification.util.Config;
-import javax.net.ssl.HttpsURLConnection;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.TrustManager;
-import javax.net.ssl.X509TrustManager;
-import javax.ws.rs.core.MediaType;
-import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
-import java.net.HttpURLConnection;
-import java.net.URL;
-import java.security.SecureRandom;
-import java.security.cert.X509Certificate;
-import java.util.*;
+import java.util.HashSet;
+import java.util.List;
public class StorageTestUtils {
- public static ClientResponse send(String path, String httpMethod, Map<String, String> headers, String requestBody,
- String query) throws Exception {
-
- log(httpMethod, StorageTestUtils.getApiPath(path + query), headers, requestBody);
- Client client = StorageTestUtils.getClient();
-
- WebResource webResource = client.resource(StorageTestUtils.getApiPath(path + query));
-
- WebResource.Builder builder = webResource.accept(MediaType.APPLICATION_JSON).type(MediaType.APPLICATION_JSON);
- headers.forEach(builder::header);
-
- return builder.method(httpMethod, ClientResponse.class, requestBody);
- }
-
- public static ClientResponse send(String url, String path, String httpMethod, Map<String, String> headers,
- String requestBody, String query) throws Exception {
-
- log(httpMethod, url + path, headers, requestBody);
- Client client = StorageTestUtils.getClient();
+ private final IStorageService storageService;
+ private final HashSet<String> allowedOps = new HashSet<>();
- WebResource webResource = client.resource(url + path);
- WebResource.Builder builder = webResource.accept(MediaType.APPLICATION_JSON).type(MediaType.APPLICATION_JSON);
- headers.forEach(builder::header);
-
- return builder.method(httpMethod, ClientResponse.class, requestBody);
+ public StorageTestUtils(DpsHeaders headers, HttpResponseBodyMapper mapper) {
+ StorageAPIConfig config = StorageAPIConfig.builder().rootUrl(Config.Instance().StorageServicePath).build();
+ StorageFactory subscriptionFactory = new StorageFactory(config, mapper);
+ storageService = subscriptionFactory.create(headers);
+ allowedOps.add("create");
+ allowedOps.add("delete");
}
- private static void log(String method, String url, Map<String, String> headers, String body) {
- System.out.println(String.format("%s: %s", method, url));
- System.out.println(body);
+ 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 StorageMessage 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);
+ storageService.upsertRecord(records);
+ // We append the index in RecordUtils to ensure the IDs are unique. Maybe we don't need to do that.
+ return new StorageMessage(records[0].getId(), KIND_TEST, "create");
}
- public static String getApiPath(String api) throws Exception {
- String baseUrl = System.getProperty("STORAGE_URL", System.getenv("STORAGE_URL"));
- URL mergedURL = new URL(baseUrl + api);
- System.out.println(mergedURL.toString());
- return mergedURL.toString();
- }
-
- protected static Client getClient() {
- TrustManager[] trustAllCerts = new TrustManager[]{new X509TrustManager() {
- @Override
- public X509Certificate[] getAcceptedIssuers() {
- return null;
- }
-
- @Override
- public void checkClientTrusted(X509Certificate[] certs, String authType) {
- }
-
- @Override
- public void checkServerTrusted(X509Certificate[] certs, String authType) {
- }
- }};
-
- try {
- SSLContext sc = SSLContext.getInstance("TLS");
- sc.init(null, trustAllCerts, new SecureRandom());
- HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
- } catch (Exception e) {
- }
- allowMethods("PATCH");
- return Client.create();
- }
-
- private static void allowMethods(String... methods) {
- try {
- Field methodsField = HttpURLConnection.class.getDeclaredField("methods");
-
- Field modifiersField = Field.class.getDeclaredField("modifiers");
- modifiersField.setAccessible(true);
- modifiersField.setInt(methodsField, methodsField.getModifiers() & ~Modifier.FINAL);
-
- methodsField.setAccessible(true);
-
- String[] oldMethods = (String[]) methodsField.get(null);
- Set<String> methodsSet = new LinkedHashSet<>(Arrays.asList(oldMethods));
- methodsSet.addAll(Arrays.asList(methods));
- String[] newMethods = methodsSet.toArray(new String[0]);
-
- methodsField.set(null/*static field*/, newMethods);
- } catch (NoSuchFieldException | IllegalAccessException e) {
- throw new IllegalStateException(e);
- }
+ public void assertFirstMessagesSimilar(StorageMessage expected, List<Message> receivedMessages) throws Exception {
+ System.out.printf("Total received messages: %d\n", receivedMessages.size());
+ StorageMessage actual = AwsTestUtils.unwrapFirst(receivedMessages, StorageMessage[].class);
+ assertTrue("Expect the op to be in the \"created\" or \"deleted\" set.", allowedOps.contains(actual.op()));
+ assertEquals(expected.kind(), actual.kind());
+ assertEquals(expected.id(), actual.id());
}
+}
- public static Map<String, String> getHeaders(String tenantName, String token, String correlationId, boolean isTestPartition) {
- Map<String, String> headers = new HashMap<>();
- if(tenantName == null || tenantName.isEmpty()) {
- tenantName = TenantUtils.getTenantName(false);
- }
- headers.put("data-partition-id", TenantUtils.getTenantName(isTestPartition));
- headers.put("Authorization", token);
+record StorageMessage(String id, String kind, String op) {
- System.out.printf("Using correlation-id for the request: %s \n", correlationId);
- headers.put("correlation-id", correlationId);
- return headers;
- }
}
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TenantUtils.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TenantUtils.java
index 92aa0406c3bfb0bd246716c358b36d7208e82740..d0427ae8c9c30e85ad63456b7ca66a464967c2dc 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TenantUtils.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TenantUtils.java
@@ -1,17 +1,17 @@
-// Copyright © Amazon Web Services
-//
-// 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.
-
+/* 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;
public class TenantUtils {
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TestNotificationsEndpoint.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TestNotificationsEndpoint.java
index d6ce07052d7ab3aa39bc00f3018ed81133122942..966157ac9bc9ac10e1614002d7b757d05adbddc3 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TestNotificationsEndpoint.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/subscriptions/TestNotificationsEndpoint.java
@@ -1,26 +1,36 @@
-// Copyright © Amazon Web Services
-//
-// 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.
-
+/* 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 com.sun.jersey.api.client.ClientResponse;
+import com.amazonaws.services.simplesystemsmanagement.AWSSimpleSystemsManagement;
+import com.amazonaws.services.simplesystemsmanagement.AWSSimpleSystemsManagementClientBuilder;
+import com.amazonaws.services.simplesystemsmanagement.model.GetParametersRequest;
+import com.amazonaws.services.simplesystemsmanagement.model.GetParametersResult;
+import com.amazonaws.services.sqs.AmazonSQS;
+import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
+import com.amazonaws.services.sqs.model.Message;
+import com.amazonaws.services.sqs.model.PurgeQueueRequest;
+import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.After;
-import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.opengroup.osdu.core.common.http.json.HttpResponseBodyMapper;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.notification.HmacSecret;
+import org.opengroup.osdu.core.common.model.notification.Secret;
import org.opengroup.osdu.core.common.model.notification.Subscription;
import org.opengroup.osdu.core.common.notification.ISubscriptionService;
import org.opengroup.osdu.core.common.notification.SubscriptionAPIConfig;
@@ -28,125 +38,393 @@ import org.opengroup.osdu.core.common.notification.SubscriptionException;
import org.opengroup.osdu.core.common.notification.SubscriptionFactory;
import org.opengroup.osdu.notification.util.*;
+import java.time.Instant;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestNotificationsEndpoint {
+
+ private static ISubscriptionService subscriptionService;
+ private static StorageTestUtils storageUtils;
+ private static LegalTagUtils legalUtils;
+ private static FileTestUtils fileUtils;
+
+ private static final int FILE_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;
+ private static final int LEGAL_STORAGE_LEGAL_QUEUE_NUM = 1;
+ private static final int REQUIRED_QUEUES = 2;
+
+ protected static final String PARTITION_TEST = "int-test-notification";
+ private static final String FILE_SERVICE_NAME = "file";
+ 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 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());
+ private static final String STORAGE_LEGAL_TAG_NAME = String.format("%s-legal-storage-%d", TenantUtils.getTenantName(true), System.currentTimeMillis());
+
+ private static AmazonSQS sqsClient;
+
+ private static final Map<String, String> serviceTopicLookup = new HashMap<>();
+
+ private final static ObjectMapper objectMapper = new ObjectMapper();
+
+ private static final int MAX_RETRIES = 5;
+ private static final int MAX_POLL_TIME = 120;
+
-public class TestNotificationsEndpoint extends TestBase {
+ private static class PushQueue {
+ private final String queueUrl;
+ private final String endpointUrl;
+ private final int queueNum;
+ private final Set<String> subscriptions = new HashSet<>();
+
+ private final List<Message> messages = new ArrayList<>();
+
+ public PushQueue(String queueUrl, String endpointUrl, int queueNum) {
+ this.queueUrl = queueUrl;
+ this.endpointUrl = endpointUrl;
+ this.queueNum = queueNum;
+ }
+
+ public void register(String serviceName) throws SubscriptionException, InterruptedException {
+ String osduTopic = serviceTopicLookup.get(serviceName);
+ assertNotEquals(null, osduTopic);
+
+ Subscription subscription = new Subscription();
+ subscription.setPushEndpoint(endpointUrl);
+ subscription.setSecret(subscriptionSecret);
+ subscription.setId(String.format("%s-%d", osduTopic, queueNum));
+ subscription.setName(String.format("%s on queue %d", osduTopic, queueNum));
+ subscription.setDescription(String.format("Testing service %s osduTopic by using push endpoint to queue %d.", serviceName, queueNum));
+ subscription.setTopic(osduTopic);
+ String subscriptionId = null;
+ int i = 0;
+ boolean errored = true;
+ while (errored) {
+ try {
+ subscriptionId = subscriptionService.create(subscription).getId();
+ i = 0;
+ errored = false;
+ } catch (SubscriptionException e) {
+ if (i >= MAX_RETRIES || e.getHttpResponse().getResponseCode() != 400)
+ throw e;
+ Thread.sleep((int) (200 * (2 << i)));
+ }
+ ++i;
+ }
+ subscriptions.add(subscriptionId);
+ }
+
+ public void pollMessages(int numMessages, int timeout) {
+ messages.clear();
+ ReceiveMessageRequest request = new ReceiveMessageRequest();
+ request.setQueueUrl(queueUrl);
+ request.setMaxNumberOfMessages(numMessages);
+ final int MAX_TIMEOUT = 20;
+ request.setWaitTimeSeconds(MAX_TIMEOUT);
+ while (messages.size() < numMessages && timeout > MAX_TIMEOUT) {
+ messages.addAll(sqsClient.receiveMessage(request).getMessages());
+ timeout -= MAX_TIMEOUT;
+ }
+ if (messages.size() < numMessages)
+ messages.addAll(sqsClient.receiveMessage(request.withWaitTimeSeconds(timeout)).getMessages());
+ }
+
+ public void pollMessages(int numMessages) {
+ pollMessages(numMessages, MAX_POLL_TIME);
+ }
- private String subscriptionId_TestPartition;
- private ISubscriptionService awssubscriptionService;
- private TestUtils testUtils;
- private static SubscriptionFactory awsfactory;
+ public List<Message> getMessages() {
+ return new ArrayList<>(messages);
+ }
+
+ public void clearLocalMessages() {
+ messages.forEach(message -> sqsClient.deleteMessage(queueUrl, message.getReceiptHandle()));
+ messages.clear();
+ }
+
+ public void removeAllKnownSubscriptions() {
+ clearLocalMessages();
+ subscriptions.forEach(TestNotificationsEndpoint::ensureSubscriptionDeleted);
+ subscriptions.clear();
+ }
+
+ public void purge() {
+ removeAllKnownSubscriptions();
+ PurgeQueueRequest purgeQueueRequest = new PurgeQueueRequest();
+ purgeQueueRequest.setQueueUrl(queueUrl);
+ sqsClient.purgeQueue(purgeQueueRequest);
+ }
+
+ public void removeAllPotentialSubscriptions(Map<String, String> allTopics) {
+ allTopics.forEach((key, value) -> {
+ String id = Base64.getEncoder().encodeToString((value + PARTITION_TEST + endpointUrl).getBytes());
+ ensureSubscriptionDeleted(id);
+ });
+ }
+ }
+
+ private record IndexerProgress(int statusCode, String[] trace, String lastUpdateTime) {
+ }
+
+ private record IndexerMessage(String id, String kind, String operationType, String status, IndexerProgress indexProgress) {
+ }
+
+
+ private static String getEnv(String envVariableName) {
+ String envVar = System.getenv(envVariableName);
+ if (envVar == null) {
+ throw new IllegalArgumentException(String.format("Must have environment variable '%s' set!", envVariableName));
+ }
+
+ return envVar;
+ }
- private static final long NOW = System.currentTimeMillis();
- private static final String RECORD_ID_PREFIX_TEST = TenantUtils.getTenantName(true) + ":query:";
- private static final String KIND_TEST = TenantUtils.getTenantName(true) + ":ds:query:1.0." + NOW;
- private static final String LEGAL_TAG = TenantUtils.getTenantName(false) + "-storage-" + System.currentTimeMillis();
- private static final String LEGAL_TAG_TEST = TenantUtils.getTenantName(true) + "-storage-" + System.currentTimeMillis();
+ private static List<PushQueue> queues;
- protected static final String PARTITION_TEST = "opendes";
+ private static Secret subscriptionSecret;
- 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";
@BeforeClass
- public static void classSetup() {
+ public static void classSetup() throws Exception {
SubscriptionAPIConfig config = SubscriptionAPIConfig.builder().rootUrl(Config.Instance().RegisterServicePath).build();
- awsfactory = new SubscriptionFactory(config);
+ SubscriptionFactory subscriptionFactory = new SubscriptionFactory(config);
+ HttpResponseBodyMapper responseBodyMapper = new HttpResponseBodyMapper(objectMapper);
+
+
+ sqsClient = AmazonSQSClientBuilder.defaultClient();
+
+ getSubscriptionSecret();
+
+ // Set up the subscription service used by this test
+ AwsTestUtils testUtils = new AwsTestUtils();
+
+ DpsHeaders dpsHeaders = testUtils.getDpsHeaders();
+ // Set up the legal service used by this test suite
+ storageUtils = new StorageTestUtils(dpsHeaders, responseBodyMapper);
+ legalUtils = new LegalTagUtils(dpsHeaders, responseBodyMapper);
+ subscriptionService = subscriptionFactory.create(dpsHeaders);
+ fileUtils = new FileTestUtils(dpsHeaders);
+
+ getTopicNames();
+ createQueues();
+ purgeQueues();
+ assertTrue(String.format("Must have at least %d Push Endpoint Queues to run this test!", REQUIRED_QUEUES), queues.size() >= REQUIRED_QUEUES);
}
- @Before
- public void setup() throws Exception {
- this.testUtils = new AwsTestUtils();
- Map<String, String> headers = new HashMap<>();
- headers.put(DpsHeaders.DATA_PARTITION_ID, TestUtils.getOsduTenant());
- headers.put(DpsHeaders.AUTHORIZATION, testUtils.getOpsToken());
- //hardcoding user here for 200 response tests. This is just initializing the subscription creation
- headers.put("x-user-id", AwsConfig.getAWSCognitoUser());
+ private static void getSubscriptionSecret() {
+ // Get the Push Endpoint Secret. Only one secret is used and shared across because otherwise, we would need to have a separate authorization for each queue, and warming up the
+ // lambda functions would need to
+ String pushEndpointSecret = getEnv("HMAC_SECRET");
+ HmacSecret hmacSecret = new HmacSecret();
+ hmacSecret.setValue(pushEndpointSecret);
+ subscriptionSecret = hmacSecret;
+ System.out.printf("TestNotificationsEndpoint.classSetup has HMAC Secret: %s\n", subscriptionSecret);
+ }
- DpsHeaders dpsHeaders = DpsHeaders.createFromMap(headers);
- awssubscriptionService = awsfactory.create(dpsHeaders);
+ private static void purgeQueues() throws InterruptedException {
+ queues.forEach(PushQueue::purge);
+ // Wait for the queues to be purged before continuing the tests.
+ Thread.sleep(60000);
+ }
- try {
- // Ensure that there is no previous subscription already registered
- // Sometimes this can happen if previous tests didn't exit properly
+ private static void createQueues() {
+ // Create the Push Endpoint Queues to ensure the right notifications are sent to the right queues.
+ String numQueuesAsString = getEnv("PUSH_ENDPOINT_NUM_QUEUES");
+ int numQueues = Integer.parseInt(numQueuesAsString);
+ queues = new ArrayList<>(numQueues);
+ for (int i = 0; i < numQueues; ++i) {
+ String endpointUrl = getEnv(String.format("PUSH_ENDPOINT_URL_%d", i));
+ String queueUrl = getEnv(String.format("PUSH_ENDPOINT_QUEUE_%d", i));
+ PushQueue queue = new PushQueue(queueUrl, endpointUrl, i);
+ queue.removeAllPotentialSubscriptions(serviceTopicLookup);
+ queues.add(queue);
+ }
+ }
- // Sadly there is no method to delete a subscription from just name, so have
- // to manually build the ID like this.
- String subscriptionId = String.format("%s%s%s",
- Config.Instance().Topic,
- PARTITION_TEST,
- Config.Instance().HMACPushUrl);
- String encodedSubscriptionId = Base64.getEncoder().encodeToString(subscriptionId.getBytes());
+ 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};
+ List<String> ssmNames = new ArrayList<>(services.length);
+ Map<String, String> parameterReverseLookup = new HashMap<>(services.length);
+ Arrays.stream(services).forEach(serviceName -> {
+ String ssmParam = String.format("/osdu/instances/%s/core/%s/osdu-topic-name", instanceName, serviceName);
+ ssmNames.add(ssmParam);
+ parameterReverseLookup.put(ssmParam, serviceName);
+ });
+ AWSSimpleSystemsManagement ssm = AWSSimpleSystemsManagementClientBuilder.defaultClient();
+ GetParametersRequest getParametersRequest = new GetParametersRequest().withNames(ssmNames);
+ GetParametersResult getParametersResult = ssm.getParameters(getParametersRequest);
+ getParametersResult.getParameters().forEach(parameter -> {
+ String serviceName = parameterReverseLookup.get(parameter.getName());
+ serviceTopicLookup.put(serviceName, parameter.getValue());
+ });
+ }
- awssubscriptionService.delete(encodedSubscriptionId);
+ private static void ensureSubscriptionDeleted(String subscriptionId) {
+ try {
+ subscriptionService.delete(subscriptionId);
+ }
+ catch (Exception e) {
+ // Don't care, because the subscription may or may not exist.
+ // If it doesn't exist, then this will erroneously error out if we don't catch it first.
}
- catch (Exception e) {}
}
@After
- @Override
public void tearDown() throws Exception {
- LegalTagUtils.delete(LEGAL_TAG_TEST, testUtils.getOpsToken(), true);
- LegalTagUtils.delete(LEGAL_TAG, testUtils.getOpsToken(), false);
- this.testUtils = null;
- }
-
- private void createResourceForTestParition() throws Exception {
- //Create a new subscription to pub/sub
- Subscription subscription = new Subscription();
- subscription.setName("Subscription-test-for-notification");
- subscription.setDescription("Subscription with test Partition for fetching notifications");
- subscription.setTopic(Config.Instance().Topic);
- subscription.setPushEndpoint(Config.Instance().HMACPushUrl);
- HmacSecret secret = new HmacSecret();
- secret.setValue(Config.Instance().hmacSecretValue);
-
- subscription.setSecret(secret);
- try {
- Subscription subscriptionCreated = awssubscriptionService.create(subscription);
- System.out.println("Subscription created successfully");
- String notificationId = subscriptionCreated.getNotificationId();
- subscriptionId_TestPartition = subscriptionCreated.getId();
- Config.Instance().NotificationId = notificationId;
- }catch (SubscriptionException e){
- System.out.println("Subscription exception inner response : " + e.getHttpResponse());
- throw e;
+ queues.forEach(PushQueue::removeAllKnownSubscriptions);
+ legalUtils.ensureDeleted(STORAGE_LEGAL_TAG_NAME);
+ legalUtils.ensureDeleted(INDEXER_LEGAL_TAG_NAME);
+ legalUtils.ensureDeleted(FILE_LEGAL_TAG_NAME);
+ Thread.sleep(10000); // Give time for the legal notifications to be sent...
+ queues.forEach(queue -> queue.removeAllPotentialSubscriptions(serviceTopicLookup));
+ purgeQueues();
+ }
+
+ private List<List<Message>> pollForMessages(int timeout, int numMessages, PushQueue... queues) throws InterruptedException {
+ List<Thread> threads = new ArrayList<>(queues.length);
+ for (PushQueue queue : queues) {
+ Runnable pollMessagesRunnable = () -> queue.pollMessages(numMessages);
+ if (timeout >= 0) {
+ pollMessagesRunnable = () -> queue.pollMessages(numMessages, timeout);
+ }
+ Thread thread = new Thread(pollMessagesRunnable);
+ thread.start();
+ threads.add(thread);
+ }
+ for (Thread thread : threads) {
+ thread.join();
}
+ List<List<Message>> messages = new ArrayList<>(queues.length);
+ for (PushQueue queue : queues) {
+ messages.add(queue.getMessages());
+ }
+ return messages;
}
- private ClientResponse createStorageRecordForTestPartition(final String correlationId) throws Exception {
- String recordId = RECORD_ID_PREFIX_TEST + UUID.randomUUID().toString();
- String jsonInput = RecordUtils.createJsonRecordWithReference(1, recordId, KIND_TEST, LEGAL_TAG_TEST, PERSISTABLE_REFERENCE, "CRS", true);
- return StorageTestUtils.send("records", "PUT", StorageTestUtils.getHeaders(TenantUtils.getTenantName(true), testUtils.getAdminToken(), correlationId, true), jsonInput, "");
+ private List<List<Message>> pollForMessages(PushQueue... queues) throws InterruptedException {
+ return pollForMessages(-1, 1, queues);
}
@Test
- public void testVerifyNotificationReceived() throws Exception {
- try {
- LegalTagUtils.create(LEGAL_TAG_TEST, testUtils.getOpsToken(), true);
- createResourceForTestParition();
- final String correlationId = UUID.randomUUID().toString();
- ClientResponse response = createStorageRecordForTestPartition(correlationId);
- assertEquals(201, response.getStatus());
- //Executing notifications response to endpoints takes an upper bound of 120s.
- Thread.sleep(120000);
-
- //Run Bash File to fetch logs from register endpoint and verify that notification was received
- String bashFileToExecute = "src/test/java/org/opengroup/osdu/notification/subscriptions/verify_register-logs.sh " + correlationId;
-
- Process process = Runtime.getRuntime().exec(bashFileToExecute);
- process.waitFor();
-
- int exitValue = process.exitValue();
- assertEquals(exitValue, 0);
- } catch (Exception e) {
- throw e;
- } finally {
- awssubscriptionService.delete(subscriptionId_TestPartition);
+ public void test_File_Notifications_Received() throws Exception {
+ synchronized (mutex) {
+ PushQueue fileQueue = queues.get(FILE_QUEUE_NUM);
+ try {
+ fileQueue.register(FILE_SERVICE_NAME);
+ legalUtils.addLegalTag(FILE_LEGAL_TAG_NAME);
+ Instant start = Instant.now();
+ String fileId = fileUtils.uploadFile(FILE_LEGAL_TAG_NAME);
+
+ // Get Queue Messages
+ List<List<Message>> messagesLists = pollForMessages(-1, 3, fileQueue);
+ assertEquals(1, messagesLists.size());
+ List<Message> fileSQSMessages = messagesLists.get(0);
+
+ fileUtils.compareMessagesWithExpected(fileId, fileSQSMessages, start);
+ } catch (Exception e) {
+ System.err.println(e.toString());
+ e.printStackTrace();
+ throw e;
+ } finally {
+ legalUtils.ensureDeleted(FILE_LEGAL_TAG_NAME);
+ }
+ }
+ }
+
+ @Test
+ public void test_Indexer_Notifications_Received() throws Exception {
+ synchronized (mutex) {
+ PushQueue storageQueue = queues.get(INDEXER_STORAGE_STORAGE_QUEUE_NUM);
+ PushQueue indexerQueue = queues.get(INDEXER_STORAGE_INDEXER_QUEUE_NUM);
+ try {
+ indexerQueue.register(INDEXER_SERVICE_NAME);
+ storageQueue.register(STORAGE_SERVICE_NAME);
+ Instant start = Instant.now();
+
+ legalUtils.addLegalTag(INDEXER_LEGAL_TAG_NAME);
+ StorageMessage expectedStorageMessage = storageUtils.createAndPutRecord(UUID.randomUUID().toString(), INDEXER_LEGAL_TAG_NAME);
+
+ // Get Queue Messages
+ List<List<Message>> messagesLists = pollForMessages(600, 1, storageQueue, indexerQueue);
+ assertEquals(2, messagesLists.size());
+ List<Message> storageSQSMessages = messagesLists.get(0);
+ List<Message> indexerSQSMessages = messagesLists.get(1);
+
+ Instant end = Instant.now();
+
+ storageUtils.assertFirstMessagesSimilar(expectedStorageMessage, storageSQSMessages);
+ IndexerMessage indexerMessage = AwsTestUtils.unwrapFirst(indexerSQSMessages, IndexerMessage[].class);
+ assertEquals(expectedStorageMessage.kind(), indexerMessage.kind());
+ assertEquals(expectedStorageMessage.id(), indexerMessage.id());
+ assertEquals("create", indexerMessage.operationType());
+ assertEquals("WARN", indexerMessage.status());
+ assertNotEquals(null, indexerMessage.indexProgress());
+ assertEquals(200, indexerMessage.indexProgress().statusCode());
+ Instant lastChanged = Instant.parse(indexerMessage.indexProgress().lastUpdateTime());
+ assertTrue(lastChanged.isAfter(start) && lastChanged.isBefore(end));
+ } catch (Exception e) {
+ System.err.println(e.toString());
+ e.printStackTrace();
+ throw e;
+ } finally {
+ legalUtils.ensureDeleted(INDEXER_LEGAL_TAG_NAME);
+ }
+ }
+ }
+
+ @Test
+ public void test_LegalAndStorage_Notifications_Received() throws Exception {
+ synchronized (mutex) {
+ PushQueue legalQueue = queues.get(LEGAL_STORAGE_LEGAL_QUEUE_NUM);
+ PushQueue storageQueue = queues.get(LEGAL_STORAGE_STORAGE_QUEUE_NUM);
+ try {
+ legalQueue.register(LEGAL_SERVICE_NAME);
+ storageQueue.register(STORAGE_SERVICE_NAME);
+
+ legalUtils.addLegalTag(STORAGE_LEGAL_TAG_NAME);
+
+ final String correlationId = UUID.randomUUID().toString();
+ StorageMessage expectedStorageMessage = storageUtils.createAndPutRecord(correlationId, STORAGE_LEGAL_TAG_NAME);
+ List<List<Message>> messagesLists = pollForMessages(storageQueue);
+ assertEquals(1, messagesLists.size());
+ List<Message> storageSQSMessages = messagesLists.get(0);
+ storageUtils.assertFirstMessagesSimilar(expectedStorageMessage, storageSQSMessages);
+ storageQueue.clearLocalMessages();
+
+ LegalMessage expectedLegalMessage = legalUtils.deleteLegalTag(STORAGE_LEGAL_TAG_NAME);
+
+ // Get Queue Messages
+ messagesLists = pollForMessages(storageQueue, legalQueue);
+ assertEquals(2, messagesLists.size());
+ storageSQSMessages = messagesLists.get(0);
+ List<Message> legalSQSMessages = messagesLists.get(1);
+
+ storageUtils.assertFirstMessagesSimilar(expectedStorageMessage, storageSQSMessages);
+ legalUtils.assertFirstMessagesSimilar(expectedLegalMessage, legalSQSMessages);
+ } catch (Exception e) {
+ System.err.println(e.toString());
+ e.printStackTrace();
+ throw e;
+ } finally {
+ legalUtils.ensureDeleted(STORAGE_LEGAL_TAG_NAME);
+ }
}
}
}
diff --git a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/util/AwsTestUtils.java b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/util/AwsTestUtils.java
index d2c64bab0ba45d473ed115c4ca445b2d8a7b2a83..5a2df46ba141375f8fc35b5c7c06fdc567cb24e3 100644
--- a/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/util/AwsTestUtils.java
+++ b/testing/notification-test-aws/src/test/java/org/opengroup/osdu/notification/util/AwsTestUtils.java
@@ -15,13 +15,20 @@
package org.opengroup.osdu.notification.util;
-import org.apache.commons.lang3.StringUtils;
-
-public class AwsTestUtils extends TestUtils{
+import static org.junit.Assert.assertTrue;
+import com.amazonaws.services.sqs.model.Message;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+public class AwsTestUtils extends TestUtils {
AwsCognitoClient client = new AwsCognitoClient();
+ private static final ObjectMapper objectMapper = new ObjectMapper();
@Override
public String getOpsToken() throws Exception {
@@ -31,6 +38,35 @@ public class AwsTestUtils extends TestUtils{
return opsToken;
}
+ public DpsHeaders getDpsHeaders() throws Exception {
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, TestUtils.getOsduTenant());
+ headers.put(DpsHeaders.AUTHORIZATION, this.getOpsToken());
+ //hardcoding user here for 200 response tests. This is just initializing the subscription creation
+ headers.put(DpsHeaders.USER_ID, AwsConfig.getAWSCognitoUser());
+
+ return DpsHeaders.createFromMap(headers);
+ }
+
+ public static <T> T unwrapFirstMessage(Message message, Class<T[]> classType) throws Exception {
+ T[] submessages = objectMapper.readValue(message.getBody(), classType);
+ assertTrue("Must have at least 1 sub-message!", 1 <= submessages.length);
+ return submessages[0];
+ }
+
+ public static <T> T unwrapFirst(List<Message> messages, Class<T[]> classType) throws Exception {
+ assertTrue("Must have at least 1 message!", 1 <= messages.size());
+
+ Message firstMessage = messages.get(0);
+ return unwrapFirstMessage(firstMessage, classType);
+ }
+
+ public static <T> T unwrapSingleFirst(List<Message> messages, Class<T> classType) throws Exception {
+ assertTrue("Must have at least 1 message!", 1 <= messages.size());
+
+ Message firstMessage = messages.get(0);
+ return objectMapper.readValue(firstMessage.getBody(), classType);
+ }
//These users don't have access to the API, so getting the token for NoAccessUser
@Override
@@ -40,6 +76,7 @@ public class AwsTestUtils extends TestUtils{
}
return adminToken;
}
+
//These users don't have access to the API, so getting the token for NoAccessUser
@Override
public String getEditorToken() throws Exception {
@@ -48,6 +85,7 @@ public class AwsTestUtils extends TestUtils{
}
return editorToken;
}
+
//These users don't have access to the API, so getting the token for NoAccessUser
@Override
public String getNoAccessToken() throws Exception {