WIP - porting tests to Weaver

build.sbt

@@ -38,7 +38,8 @@ lazy val core = project
       libraryDependencies ++= Seq(
         "co.fs2" %% "fs2-core" % fs2Version,
         "org.typelevel" %% "cats-effect" % catsEffectVersion,
-        "org.apache.kafka" % "kafka-clients" % kafkaVersion
+        "org.apache.kafka" % "kafka-clients" % kafkaVersion,
+        "com.disneystreaming" %% "weaver-cats" % "0.7.11" % Test
       )
     ),
     publishSettings,
@@ -364,6 +365,7 @@ lazy val scalaSettings = Seq(
 lazy val testSettings = Seq(
   Test / logBuffered := false,
   Test / parallelExecution := false,
+  testFrameworks += new TestFramework("weaver.framework.CatsEffect"),
   Test / testOptions += Tests.Argument("-oDF")
 )
 

modules/core/src/test/resources/logback-test.xml

@@ -6,6 +6,8 @@
     </encoder>
   </appender>
 
+  <logger name="org.apache.kafka.common.utils.AppInfoParser" level = "ERROR"/>
+  <logger name="org.apache.kafka.common.metrics.Metrics" level = "ERROR"/>
   <root level="INFO" additivity="false">
     <appender-ref ref="STDOUT"/>
   </root>

modules/core/src/test/scala/fs2/kafka/BaseKafkaSpec.scala

@@ -26,12 +26,12 @@ This file contains code derived from the Embedded Kafka library
  */
 package fs2.kafka
 
-import cats.effect.Sync
+import cats.effect.{IO, Sync}
 import fs2.kafka.internal.converters.collection._
 
 import java.util.UUID
 import scala.util.Failure
-import com.dimafeng.testcontainers.{ForAllTestContainer, KafkaContainer}
+import com.dimafeng.testcontainers.{ForAllTestContainer}
 import org.apache.kafka.clients.admin.AdminClientConfig
 import org.apache.kafka.clients.consumer.{KafkaConsumer => KConsumer}
 import org.apache.kafka.clients.producer.{
@@ -56,41 +56,22 @@ import org.apache.kafka.common.serialization.StringSerializer
 import java.util.concurrent.TimeUnit
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.scalatest.Args
+import weaver.Expectations
 
-abstract class BaseKafkaSpec extends BaseAsyncSpec with ForAllTestContainer {
+abstract class BaseKafkaSpec extends BaseAsyncSpec with ForAllTestContainer with BaseKafkaSpecBase {
+
+  override def runTest(testName: String, args: Args) = super.runTest(testName, args)
+}
+
+trait BaseKafkaSpecBase {
+  lazy val container = BaseWeaverSpecShared.container
 
   final val adminClientCloseTimeout: FiniteDuration = 2.seconds
   final val transactionTimeoutInterval: FiniteDuration = 1.second
 
   final val consumerPollingTimeout: FiniteDuration = 1.second
   protected val producerPublishTimeout: FiniteDuration = 10.seconds
 
-  override def runTest(testName: String, args: Args) = super.runTest(testName, args)
-
-  private val imageVersion = "7.0.1"
-
-  private lazy val imageName = Option(System.getProperty("os.arch")) match {
-    case Some("aarch64") =>
-      "niciqy/cp-kafka-arm64" // no official docker image for ARM is available yet
-    case _ => "confluentinc/cp-kafka"
-  }
-
-  override val container: KafkaContainer = new KafkaContainer()
-    .configure { container =>
-      container
-        .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
-        .withEnv(
-          "KAFKA_TRANSACTION_ABORT_TIMED_OUT_TRANSACTION_CLEANUP_INTERVAL_MS",
-          transactionTimeoutInterval.toMillis.toString
-        )
-        .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
-        .withEnv("KAFKA_AUTHORIZER_CLASS_NAME", "kafka.security.authorizer.AclAuthorizer")
-        .withEnv("KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND", "true")
-        .setDockerImageName(s"$imageName:$imageVersion")
-
-      ()
-    }
-
   implicit final val stringSerializer: KafkaSerializer[String] = new StringSerializer
 
   implicit final val stringDeserializer: KafkaDeserializer[String] = new StringDeserializer
@@ -152,6 +133,11 @@ abstract class BaseKafkaSpec extends BaseAsyncSpec with ForAllTestContainer {
   final def withTopic[A](f: String => A): A =
     f(nextTopicName())
 
+  final def withTopic(partitions: Int)(f: String => IO[Expectations]): IO[Expectations] =
+    IO(nextTopicName()).flatMap { topic =>
+      IO.blocking(createCustomTopic(topic, partitions = partitions)) >> f(topic)
+    }
+
   final def withKafkaConsumer(
     nativeSettings: Map[String, AnyRef]
   ): WithKafkaConsumer =

modules/core/src/test/scala/fs2/kafka/BaseWeaverSpec.scala

@@ -0,0 +1,74 @@
+/*
+This file contains code derived from the Embedded Kafka library
+(https://github.com/embeddedkafka/embedded-kafka), the license for which is reproduced below.
+
+   The MIT License (MIT)
+
+   Copyright (c) 2016 Emanuele Blanco
+
+   Permission is hereby granted, free of charge, to any person obtaining a copy
+   of this software and associated documentation files (the "Software"), to deal
+   in the Software without restriction, including without limitation the rights
+   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+   copies of the Software, and to permit persons to whom the Software is
+   furnished to do so, subject to the following conditions:
+
+   The above copyright notice and this permission notice shall be included in all
+   copies or substantial portions of the Software.
+
+   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+   SOFTWARE.
+ */
+package fs2.kafka
+
+import cats.effect.{IO, Resource}
+import com.dimafeng.testcontainers.KafkaContainer
+import weaver.{GlobalResource, GlobalWrite, IOSuite}
+
+import scala.concurrent.duration.{DurationInt, FiniteDuration}
+
+object BaseWeaverSpecShared extends GlobalResource {
+  protected val imageVersion = "7.0.1"
+
+  protected val imageName = Option(System.getProperty("os.arch")) match {
+    case Some("aarch64") =>
+      "niciqy/cp-kafka-arm64" // no official docker image for ARM is available yet
+    case _ => "confluentinc/cp-kafka"
+  }
+  final val transactionTimeoutInterval: FiniteDuration = 1.second
+
+  lazy val container: KafkaContainer = makeContainer()
+
+  def makeContainer() =
+    new KafkaContainer()
+      .configure { container =>
+        container
+          .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+          .withEnv(
+            "KAFKA_TRANSACTION_ABORT_TIMED_OUT_TRANSACTION_CLEANUP_INTERVAL_MS",
+            transactionTimeoutInterval.toMillis.toString
+          )
+          .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
+          .withEnv("KAFKA_AUTHORIZER_CLASS_NAME", "kafka.security.authorizer.AclAuthorizer")
+          .withEnv("KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND", "true")
+          .setDockerImageName(s"$imageName:$imageVersion")
+
+        ()
+      }
+
+  override def sharedResources(global: GlobalWrite): Resource[IO, Unit] =
+    ContainerResource(IO(container)).evalMap(global.put(_))
+}
+
+abstract class BaseWeaverSpec extends IOSuite with BaseKafkaSpecBase {
+
+  override type Res = KafkaContainer
+
+  override def sharedResource: Resource[IO, KafkaContainer] = ContainerResource(IO(container))
+
+}

modules/core/src/test/scala/fs2/kafka/ContainerResource.scala

@@ -0,0 +1,12 @@
+package fs2.kafka
+
+import cats.effect.{Resource, Sync}
+import cats.syntax.all._
+import com.dimafeng.testcontainers.Container
+
+object ContainerResource {
+  def apply[F[_], C <: Container](container: F[C])(implicit F: Sync[F]): Resource[F, C] =
+    Resource.make(container.flatTap { container =>
+      F.blocking(container.start())
+    })(c => F.blocking(c.stop()))
+}