Add more specs

aws/sqs/integration/src/test/scala/com/commercetools/queue/sqs/SqsClientSuite.scala

@@ -17,20 +17,43 @@
 package com.commercetools.queue.sqs
 
 import cats.effect.{IO, Resource}
+import cats.syntax.all._
 import com.commercetools.queue.QueueClient
 import com.commercetools.queue.aws.sqs.SQSClient
 import com.commercetools.queue.testkit.QueueClientSuite
-import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider
+import software.amazon.awssdk.auth.credentials.{AnonymousCredentialsProvider, AwsBasicCredentials, StaticCredentialsProvider}
 import software.amazon.awssdk.regions.Region
 
 import java.net.URI
+import scala.jdk.CollectionConverters.CollectionHasAsScala
 
 class SqsClientSuite extends QueueClientSuite {
 
+  private def config =
+    booleanOrDefault("AWS_SQS_USE_EMULATOR", default = true).ifM(
+      ifTrue =
+        IO.pure((Region.EU_WEST_1, AnonymousCredentialsProvider.create(), Some(new URI("http://localhost:4566")))),
+      ifFalse = for {
+        awsRegion <- string("AWS_SQS_REGION")
+        region <- Region
+          .regions()
+          .asScala
+          .find(_.id == awsRegion)
+          .liftTo[IO](new IllegalArgumentException(s"Cannot find any suitable AWS region from $awsRegion value!"))
+        credentials <- (string("AWS_SQS_ACCESS_KEY"), string("AWS_SQS_ACCESS_SECRET")).mapN((accessKey, accessSecret) =>
+          StaticCredentialsProvider.create(
+            AwsBasicCredentials.create(accessKey, accessSecret)
+          ))
+      } yield (region, credentials, None)
+    )
+
   override def client: Resource[IO, QueueClient[IO]] =
-    SQSClient[IO](
-      Region.EU_WEST_1,
-      AnonymousCredentialsProvider.create(),
-      endpoint = Some(new URI("http://localhost:4566")))
+    config.toResource.flatMap { case (region, credentials, endpoint) =>
+      SQSClient[IO](
+        region,
+        credentials,
+        endpoint = endpoint
+      )
+    }
 
 }

azure/service-bus/integration/README.md

@@ -0,0 +1,9 @@
+# How to run tests
+
+Tests are using [AzureCliCredential](https://learn.microsoft.com/en-us/java/api/com.azure.identity.azureclicredential?view=azure-java-stable).
+Make sure to be assigned with the [Azure Service Bus Data Owner](https://learn.microsoft.com/en-us/azure/role-based-access-control/built-in-roles/integration#azure-service-bus-data-owner) role.
+
+Steps:
+- `az login`
+- `export AZURE_SERVICEBUS_HOSTNAME=<your-servicebus-hostname>`
+- `sbt "project azureServiceBusIt" test`

azure/service-bus/integration/src/test/scala/com/commercetools/queue/servicebus/ServiceBusClientSuite.scala

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2024 Commercetools GmbH
+ *
+ * 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 com.commercetools.queue.servicebus
+
+import cats.effect.{IO, Resource}
+import com.azure.identity.AzureCliCredentialBuilder
+import com.commercetools.queue.QueueClient
+import com.commercetools.queue.azure.servicebus.ServiceBusClient
+import com.commercetools.queue.testkit.QueueClientSuite
+
+class ServiceBusClientSuite extends QueueClientSuite {
+
+  private def config = string("AZURE_SERVICEBUS_HOSTNAME")
+  override val inFlightMessagesStatsSupported: Boolean = false // not supported
+
+  override def client: Resource[IO, QueueClient[IO]] =
+    config.toResource.flatMap { namespace =>
+      ServiceBusClient[IO](
+        namespace = namespace,
+        credentials = new AzureCliCredentialBuilder().build()
+      )
+    }
+}

build.sbt

@@ -84,7 +84,8 @@ lazy val testkit = crossProject(JVMPlatform)
     name := "fs2-queues-testkit",
     libraryDependencies ++= List(
       "org.scalameta" %%% "munit" % Versions.munit,
-      "org.typelevel" %%% "munit-cats-effect-3" % Versions.munitCatsEffect
+      "org.typelevel" %%% "munit-cats-effect-3" % Versions.munitCatsEffect,
+      "org.slf4j" % "slf4j-simple" % "2.0.16"
     )
   )
   .dependsOn(core)
@@ -142,6 +143,17 @@ lazy val azureServiceBus = crossProject(JVMPlatform)
   )
   .dependsOn(core, testkit % Test)
 
+lazy val azureServiceBusIt = project
+  .in(file("azure/service-bus/integration"))
+  .enablePlugins(NoPublishPlugin)
+  .settings(commonSettings)
+  .settings(
+    libraryDependencies ++= List(
+      "com.azure" % "azure-identity" % "1.11.1"
+    )
+  )
+  .dependsOn(azureServiceBus.jvm % Test, testkit.jvm % Test)
+
 lazy val awsSQS = crossProject(JVMPlatform)
   .crossType(CrossType.Pure)
   .in(file("aws/sqs"))

core/src/main/scala/com/commercetools/queue/QueuePublisher.scala

@@ -30,7 +30,7 @@ sealed abstract class QueuePublisher[F[_], T](implicit F: MonadCancel[F, Throwab
   def queueName: String
 
   /**
-   * Returns a way to bush messages into the queue.
+   * Returns a way to push messages into the queue.
    * This is a low-level construct, mainly aiming at integrating existing
    * code bases that require to push explicitly.
    *

gcp/pubsub/integration/src/test/scala/com/commercetools/queue/pubsub/PubSubClientSuite.scala

@@ -24,9 +24,27 @@ import com.google.api.gax.core.NoCredentialsProvider
 
 class PubSubClientSuite extends QueueClientSuite {
 
-  override val queueUpdateSupported = false
+  private def isEmulatorDefault = true
+  private def isEmulatorEnvVar = "GCP_PUBSUB_USE_EMULATOR"
+
+  override val queueUpdateSupported: Boolean = false // not supported
+  override val inFlightMessagesStatsSupported: Boolean = false // not supported
+  override val delayedMessagesStatsSupported: Boolean = false // not supported
+  override val messagesStatsSupported: Boolean = // // not supported in the emulator
+    !sys.env.get(isEmulatorEnvVar).map(_.toBoolean).getOrElse(isEmulatorDefault)
+
+  private def config =
+    booleanOrDefault(isEmulatorEnvVar, default = isEmulatorDefault).ifM(
+      ifTrue = IO.pure(("test-project", NoCredentialsProvider.create(), Some("localhost:8042"))),
+      ifFalse = for {
+        project <- string("GCP_PUBSUB_PROJECT")
+        credentials = NoCredentialsProvider.create() // TODO
+      } yield (project, credentials, None)
+    )
 
   override def client: Resource[IO, QueueClient[IO]] =
-    PubSubClient("test-project", NoCredentialsProvider.create(), endpoint = Some("localhost:8042"))
+    config.toResource.flatMap { case (project, credentials, endpoint) =>
+      PubSubClient(project, credentials, endpoint = endpoint)
+    }
 
 }

gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/PubSubPublisher.scala

@@ -39,10 +39,10 @@ private class PubSubPublisher[F[_], T](
       .fromAutoCloseable {
         F.blocking {
           val builder =
-            PublisherStubSettings.newBuilder()
-          builder
-            .setCredentialsProvider(credentials)
-            .setTransportChannelProvider(channelProvider)
+            PublisherStubSettings
+              .newBuilder()
+              .setCredentialsProvider(credentials)
+              .setTransportChannelProvider(channelProvider)
           endpoint.foreach(builder.setEndpoint(_))
           GrpcPublisherStub.create(builder.build())
         }

gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/PubSubStatistics.scala

@@ -40,7 +40,7 @@ private class PubSubStatistics[F[_]](
             .newBuilder()
             .setCredentialsProvider(credentials)
             .setTransportChannelProvider(channelProvider)
-          endpoint.foreach(builder.setEndpoint(_))
+          endpoint.foreach(builder.setEndpoint)
           GrpcMetricServiceStub.create(builder.build())
         }
       }

testkit/src/main/scala/com/commercetools/queue/testkit/QueueAdministrationSuite.scala

@@ -0,0 +1,40 @@
+package com.commercetools.queue.testkit
+
+import com.commercetools.queue.QueueConfiguration
+import munit.CatsEffectSuite
+
+import scala.concurrent.duration._
+
+/**
+ * This suite tests that the features of a [[com.commercetools.queue.QueueAdministration QueueAdministration]] are properly
+ * implemented for a concrete client.
+ */
+trait QueueAdministrationSuite extends CatsEffectSuite { self: QueueClientSuite =>
+
+  withQueue.test("existing queue should be indicated as such") { queueName =>
+    val client = clientFixture()
+    assertIO(client.administration.exists(queueName), true)
+  }
+
+  test("non existing queue should be indicated as such") {
+    val client = clientFixture()
+    assertIO(client.administration.exists("not-existing"), false)
+  }
+
+  withQueue.test("get configuration") { queueName =>
+    val admin = clientFixture().administration
+    assertIO(admin.configuration(queueName), QueueConfiguration(originalMessageTTL, originalLockTTL))
+  }
+
+  withQueue.test("configuration can be updated") { queueName =>
+    assume(queueUpdateSupported, "The test environment does not support queue update")
+    val admin = clientFixture().administration
+    for {
+      _ <- admin.update(queueName, Some(originalMessageTTL + 1.minute), Some(originalLockTTL + 10.seconds))
+      _ <- assertIO(
+        admin.configuration(queueName),
+        QueueConfiguration(originalMessageTTL + 1.minute, originalLockTTL + 10.seconds))
+    } yield ()
+  }
+
+}