This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 293db00f0 Replace LocalStack with DynamoDB local and GoAws in aws-spi
tests (#1901)
293db00f0 is described below
commit 293db00f063ce9362d5dda9deb2eb06230029ba1
Author: Adarsh Mishra <[email protected]>
AuthorDate: Thu Sep 3 00:56:32 2026 +0530
Replace LocalStack with DynamoDB local and GoAws in aws-spi tests (#1901)
Co-authored-by: Adarsh <[email protected]>
---
.../connectors/awsspi/BaseAwsClientTest.scala | 24 +++++++----
.../connectors/awsspi/dynamodb/TestDynamoDB.scala | 6 +--
.../stream/connectors/awsspi/sns/TestSNS.scala | 6 +--
.../LocalStackReadyLogWaitStrategy.scala | 48 ----------------------
4 files changed, 21 insertions(+), 63 deletions(-)
diff --git
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/BaseAwsClientTest.scala
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/BaseAwsClientTest.scala
index d07e958f5..00763e8cc 100644
---
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/BaseAwsClientTest.scala
+++
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/BaseAwsClientTest.scala
@@ -20,13 +20,14 @@ package org.apache.pekko.stream.connectors.awsspi
import java.net.URI
import org.apache.pekko
-import
pekko.stream.connectors.awsspi.testcontainers.LocalStackReadyLogWaitStrategy
+import pekko.stream.connectors.awsspi.testcontainers.TimeoutWaitStrategy
import com.dimafeng.testcontainers.{ ForAllTestContainer, GenericContainer }
import org.scalatest.concurrent.{ Eventually, Futures, IntegrationPatience }
import org.scalatest.BeforeAndAfter
import software.amazon.awssdk.core.SdkClient
import software.amazon.awssdk.regions.Region
+import scala.concurrent.duration._
import scala.util.Random
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
@@ -49,17 +50,26 @@ trait BaseAwsClientTest[C <: SdkClient]
def randomIdentifier(length: Int): String =
Random.alphanumeric.take(length).mkString
}
-trait LocalstackBaseAwsClientTest[C <: SdkClient] extends BaseAwsClientTest[C]
{
- def service: String
+trait DynamoDBLocalBaseAwsClientTest[C <: SdkClient] extends
BaseAwsClientTest[C] {
+
+ lazy val exposedServicePort: Int = 8000
+
+ override lazy val container: GenericContainer =
+ new GenericContainer(
+ dockerImage = "amazon/dynamodb-local:3.3.1",
+ exposedPorts = Seq(exposedServicePort),
+ waitStrategy = Some(TimeoutWaitStrategy(10.seconds)))
+}
+
+trait GoAwsSNSBaseAwsClientTest[C <: SdkClient] extends BaseAwsClientTest[C] {
- lazy val exposedServicePort: Int = 4566
+ lazy val exposedServicePort: Int = 4100
override lazy val container: GenericContainer =
new GenericContainer(
- dockerImage = "localstack/localstack:4.14",
+ dockerImage = "pafortin/goaws:v0.3.1",
exposedPorts = Seq(exposedServicePort),
- env = Map("SERVICES" -> service),
- waitStrategy = Some(LocalStackReadyLogWaitStrategy))
+ waitStrategy = Some(TimeoutWaitStrategy(10.seconds)))
}
trait ElasticMQSQSBaseAwsClientTest[C <: SdkClient] extends
BaseAwsClientTest[C] {
diff --git
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/dynamodb/TestDynamoDB.scala
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/dynamodb/TestDynamoDB.scala
index 42c758286..f680d357a 100644
---
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/dynamodb/TestDynamoDB.scala
+++
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/dynamodb/TestDynamoDB.scala
@@ -18,12 +18,12 @@
package org.apache.pekko.stream.connectors.awsspi.dynamodb
import org.apache.pekko
-import pekko.stream.connectors.awsspi.{ LocalstackBaseAwsClientTest,
PekkoHttpAsyncHttpService }
+import pekko.stream.connectors.awsspi.{ DynamoDBLocalBaseAwsClientTest,
PekkoHttpAsyncHttpService }
import software.amazon.awssdk.auth.credentials.{ AwsBasicCredentials,
StaticCredentialsProvider }
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient
import software.amazon.awssdk.services.dynamodb.model._
-class TestDynamoDB extends LocalstackBaseAwsClientTest[DynamoDbAsyncClient] {
+class TestDynamoDB extends DynamoDBLocalBaseAwsClientTest[DynamoDbAsyncClient]
{
"DynamoDB" should {
"create a table" in withClient { implicit client =>
val attributes =
AttributeDefinition.builder.attributeName("film_id").attributeType(ScalarAttributeType.S).build()
@@ -73,6 +73,4 @@ class TestDynamoDB extends
LocalstackBaseAwsClientTest[DynamoDbAsyncClient] {
}
}
- override val service: String = "dynamodb"
-
}
diff --git
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/sns/TestSNS.scala
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/sns/TestSNS.scala
index d73fcb849..d8adf03c5 100644
---
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/sns/TestSNS.scala
+++
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/sns/TestSNS.scala
@@ -18,12 +18,12 @@
package org.apache.pekko.stream.connectors.awsspi.sns
import org.apache.pekko
-import pekko.stream.connectors.awsspi.{ LocalstackBaseAwsClientTest,
PekkoHttpAsyncHttpService }
+import pekko.stream.connectors.awsspi.{ GoAwsSNSBaseAwsClientTest,
PekkoHttpAsyncHttpService }
import software.amazon.awssdk.auth.credentials.{ AwsBasicCredentials,
StaticCredentialsProvider }
import software.amazon.awssdk.services.sns.SnsAsyncClient
import software.amazon.awssdk.services.sns.model.{ CreateTopicRequest,
PublishRequest }
-class TestSNS extends LocalstackBaseAwsClientTest[SnsAsyncClient] {
+class TestSNS extends GoAwsSNSBaseAwsClientTest[SnsAsyncClient] {
"Async SNS client" should {
"publish a message to a topic" in withClient { implicit client =>
@@ -53,6 +53,4 @@ class TestSNS extends
LocalstackBaseAwsClientTest[SnsAsyncClient] {
client.close()
}
}
-
- override def service: String = "sns"
}
diff --git
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/testcontainers/LocalStackReadyLogWaitStrategy.scala
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/testcontainers/LocalStackReadyLogWaitStrategy.scala
deleted file mode 100644
index 07ed0490c..000000000
---
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/testcontainers/LocalStackReadyLogWaitStrategy.scala
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.pekko.stream.connectors.awsspi.testcontainers
-
-import java.util.concurrent.{ TimeUnit, TimeoutException }
-import java.util.function.Predicate
-
-import org.testcontainers.DockerClientFactory
-import org.testcontainers.containers.ContainerLaunchException
-import org.testcontainers.containers.output.{ OutputFrame, WaitingConsumer }
-import org.testcontainers.containers.wait.strategy.AbstractWaitStrategy
-import org.testcontainers.utility.LogUtils
-
-/**
- * This strategy is based on the container log "Ready." from Localstack. Once
it's printed out, the container is good
- * to go.
- */
-object LocalStackReadyLogWaitStrategy extends AbstractWaitStrategy {
- override def waitUntilReady(): Unit = {
- val waitingConsumer = new WaitingConsumer
- LogUtils.followOutput(DockerClientFactory.instance.client,
waitStrategyTarget.getContainerId, waitingConsumer)
-
- val waitPredicate: Predicate[OutputFrame] = (outputFrame: OutputFrame) =>
- outputFrame.getUtf8String.contains("Ready.")
-
- try
- waitingConsumer.waitUntil(waitPredicate, startupTimeout.getSeconds,
TimeUnit.SECONDS, 1)
- catch {
- case _: TimeoutException =>
- throw new ContainerLaunchException("Timed out waiting for localstack")
- }
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
