Migrate CustomServerContextTest and SqlQueuePersistenceServer to new test structure, updating to AWS SDK v2 clients

Author: micossow

rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/CustomServerContextTest.scala

@@ -0,0 +1,67 @@
+package org.elasticmq.rest.sqs.aws
+
+import org.elasticmq.NodeAddress
+import org.elasticmq.rest.sqs.client.{AwsSdkV2SqsClient, SqsClient}
+import org.elasticmq.rest.sqs.{SQSRestServer, SQSRestServerBuilder}
+import org.scalatest.BeforeAndAfter
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.sqs.{SqsClient => AwsSqsClient}
+
+import java.net.URI
+import scala.util.Try
+
+class CustomServerContextTests extends AmazonJavaSdkNewTestBase with BeforeAndAfter {
+
+  var testClient: SqsClient = _
+  var clientV2: AwsSqsClient = _
+  var server: SQSRestServer = _
+
+  val awsAccountId = "123456789012"
+  val awsRegion = "elasticmq"
+  val port = 9324
+  val contextPath = "xyz"
+
+  before {
+    server = SQSRestServerBuilder
+      .withPort(port)
+      .withServerAddress(NodeAddress(port = port, contextPath = contextPath))
+      .withAWSAccountId(awsAccountId)
+      .withAWSRegion(awsRegion)
+      .start()
+
+    server.waitUntilStarted()
+
+    clientV2 = AwsSqsClient
+      .builder()
+      .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
+      .region(Region.US_EAST_1)
+      .endpointOverride(new URI(s"http://localhost:$port/$contextPath"))
+      .build()
+
+    testClient = new AwsSdkV2SqsClient(clientV2)
+  }
+
+  after {
+    clientV2.close()
+    Try(server.stopAndWait())
+  }
+
+  test("should create queue when context path is set") {
+    val queueUrl = testClient.createQueue("testQueue").toOption.get
+    queueUrl should be("http://localhost:9324/xyz/123456789012/testQueue")
+  }
+
+  test("should send and receive messages when context path is set") {
+    val queueUrl = testClient.createQueue("testQueue").toOption.get
+
+    testClient.sendMessage("http://localhost:9324/xyz/123456789012/testQueue", "test msg1").isRight shouldBe true
+    testClient.sendMessage("http://localhost:9324/xyz/queue/testQueue", "test msg2").isRight shouldBe true
+    testClient.sendMessage("http://localhost:9324/xyz/abc/testQueue", "test msg3").isRight shouldBe true
+    testClient.sendMessage("http://localhost:9324/xyz/testQueue", "test msg4").isRight shouldBe true
+
+    val messages = testClient.receiveMessage(queueUrl, maxNumberOfMessages = Some(10))
+
+    messages.map(_.body) should contain allOf ("test msg1", "test msg2", "test msg3", "test msg4")
+  }
+}

rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/SqlQueuePersistenceServer.scala

@@ -1,29 +1,102 @@
 package org.elasticmq.rest.sqs.aws
 
+import org.apache.pekko.actor.{ActorRef, ActorSystem, Props}
+import org.apache.pekko.util.Timeout
+import org.elasticmq.actor.QueueManagerActor
+import org.elasticmq.actor.queue.QueueEvent
 import org.elasticmq.actor.reply._
-import org.elasticmq.persistence.sql.GetAllMessages
-import org.elasticmq.rest.sqs.SqlQueuePersistenceServer
+import org.elasticmq.persistence.sql.{GetAllMessages, SqlQueuePersistenceActor, SqlQueuePersistenceConfig}
 import org.elasticmq.rest.sqs.client._
 import org.elasticmq.rest.sqs.model.RedrivePolicy
 import org.elasticmq.rest.sqs.model.RedrivePolicyJson.format
-import org.elasticmq.util.Logging
+import org.elasticmq.rest.sqs.{SQSRestServer, SQSRestServerBuilder}
+import org.elasticmq.util.{Logging, NowProvider}
+import org.elasticmq.{NodeAddress, StrictSQSLimits}
 import org.scalatest.BeforeAndAfter
+import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
+import org.scalatest.time.{Seconds, Span}
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.sqs.{SqsClient => AwsSqsClient}
 import spray.json.enrichAny
 
+import java.net.URI
+import scala.util.Try
+
 class SqlQueuePersistenceTests
     extends AnyFunSuite
-    with SqlQueuePersistenceServer
+    with ScalaFutures
     with BeforeAndAfter
     with Matchers
     with Logging {
 
+  private val awsAccountId = "123456789012"
+  private val awsRegion = "elasticmq"
+
+  private val actorSystem: ActorSystem = ActorSystem("elasticmq-test-v2")
+  private var strictServer: SQSRestServer = _
+
+  var clientV2: AwsSqsClient = _
+  var testClient: AwsSdkV2SqsClient = _
+  var store: ActorRef = _
+
+  implicit val timeout: Timeout = {
+    import scala.concurrent.duration._
+    Timeout(5.seconds)
+  }
+
+  implicit val defaultPatience: PatienceConfig = PatienceConfig(timeout = Span(5, Seconds))
+
   val testQueueName = "testQueue1"
   val deadLetterQueueName = "testDLQ1"
 
-  val awsAccountId = "123456789012"
-  val awsRegion = "elasticmq"
+  def startServerAndRun(pruneDataOnInit: Boolean)(body: => Unit): Unit = {
+    startServerAndSetupClient(pruneDataOnInit)
+    try {
+      body
+    } finally {
+      stopServerAndClient()
+    }
+  }
+
+  private def startServerAndSetupClient(pruneDataOnInit: Boolean): Unit = {
+    val persistenceConfig = SqlQueuePersistenceConfig(
+      enabled = true,
+      driverClass = "org.h2.Driver",
+      uri = "jdbc:h2:./elasticmq-h2-v2",
+      pruneDataOnInit = pruneDataOnInit
+    )
+
+    store = actorSystem.actorOf(Props(new SqlQueuePersistenceActor(persistenceConfig, List.empty)))
+    val manager = actorSystem.actorOf(Props(new QueueManagerActor(new NowProvider(), StrictSQSLimits, Some(store))))
+
+    strictServer = SQSRestServerBuilder
+      .withActorSystem(actorSystem)
+      .withQueueManagerActor(manager)
+      .withPort(9323) // different port to avoid conflicts
+      .withServerAddress(NodeAddress(port = 9323))
+      .withAWSAccountId(awsAccountId)
+      .withAWSRegion(awsRegion)
+      .start()
+
+    (store ? QueueEvent.Restore(manager)).futureValue
+
+    clientV2 = AwsSqsClient
+      .builder()
+      .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
+      .region(Region.EU_CENTRAL_1)
+      .endpointOverride(new URI("http://localhost:9323"))
+      .build()
+
+    testClient = new AwsSdkV2SqsClient(clientV2)
+  }
+
+  private def stopServerAndClient(): Unit = {
+    if (clientV2 != null) clientV2.close()
+    if (strictServer != null) Try(strictServer.stopAndWait())
+  }
 
   test("should persist the messages and after restart read the messages") {
     startServerAndRun(pruneDataOnInit = true) {