Add ControlledTestKit for simulated-time actor tests

README.md

@@ -2637,6 +2637,65 @@ object ShardingApp extends IOApp {
 
 - Initializes the actor system and sends a series of Shard messages to demonstrate actor creation and message processing.
 
+## Testing Actors
+
+Cats-Actors ships a companion `cats-actors-testkit` module with helpers for writing deterministic actor tests. Add it to your build (test scope):
+
+```scala
+// JVM / Scala.js / Scala Native (use %%% for cross-platform)
+libraryDependencies += "com.github.cloudmark.cats-actors" %%% "cats-actors-testkit" % "2.1.0" % Test
+```
+
+### Controlled (simulated) time with `ControlledTestKit`
+
+`ControlledTestKit` runs your test logic inside Cats Effect's `TestControl`, so **simulated time is advanced for you** — scheduled work and timeouts resolve instantly instead of forcing your suite to sleep in real time. Mix the trait into your spec and wrap the test body in `withActorSystemIO`, which provisions an `ActorSystem[IO]`, ticks the simulated clock (90 seconds by default), and yields the result.
+
+The returned `IO` simply *succeeds with your value* or *fails the effect* if the test errors, is canceled, or does not finish within the time window (the latter two surface as a `ControlledTestException`). It is test-framework agnostic — any runner that can execute an `IO` works. To inspect a failure instead of propagating it, call `.attempt` on the result.
+
+```scala
+import cats.effect.{IO, Ref}
+import com.suprnation.actor.Actor.{Actor, Receive}
+import com.suprnation.actor.test.ControlledTestKit
+import cats.effect.testing.scalatest.AsyncIOSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AsyncWordSpec
+
+import scala.concurrent.duration._
+
+class SchedulerSpec
+    extends AsyncWordSpec
+    with AsyncIOSpec
+    with Matchers
+    with ControlledTestKit {
+
+  "An actor scheduling delayed work" should {
+    "observe the task firing without waiting in real time" in
+      withActorSystemIO { system =>
+        for {
+          fired <- Ref[IO].of(false)
+          _ <- system.actorOf(
+            new Actor[IO, String] {
+              override def preStart: IO[Unit] =
+                // scheduleOnce_ fires in the background (it `.start`s a fiber);
+                // scheduleOnce would instead block preStart for the full delay.
+                context.system.scheduler.scheduleOnce_(5.seconds)(fired.set(true)).void
+
+              override def receive: Receive[IO, String] = { case _ => IO.unit }
+            },
+            "scheduler-actor"
+          )
+          // Written as a 6s wait, but ControlledTestKit advances simulated
+          // time, so the test still completes instantly.
+          _ <- IO.sleep(6.seconds)
+          result <- fired.get
+        } yield result shouldBe true
+      }
+  }
+}
+```
+
+`ControlledTestKit` extends `TestKit`, so all the message-assertion helpers (`expectMsgs`, `expectMsgType`, `awaitTerminated`, test probes, …) are available alongside the controlled-time runner.
+
 ## Looking for Another Example?
 
 If you need another example or have a specific scenario in mind, please open an issue on our GitHub repository. We'll

build.sbt

@@ -79,8 +79,33 @@ lazy val catsActorsJVM = catsActors.jvm
 lazy val catsActorsJS = catsActors.js
 lazy val catsActorsNative = catsActors.native
 
+lazy val testkit = crossProject(JSPlatform, JVMPlatform, NativePlatform)
+  .crossType(CrossType.Full)
+  .in(file("testkit"))
+  .dependsOn(catsActors)
+  .settings(commonSettings)
+  .settings(
+    name := "cats-actors-testkit",
+    libraryDependencies ++= Seq(
+      "org.typelevel" %%% "cats-effect-testkit" % "3.7.0",
+      "org.scalatest" %%% "scalatest" % "3.2.18" % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % "1.8.0" % Test
+    )
+  )
+
+lazy val testkitJVM = testkit.jvm
+lazy val testkitJS = testkit.js
+lazy val testkitNative = testkit.native
+
 lazy val root = (project in file("."))
-  .aggregate(catsActorsJVM, catsActorsJS, catsActorsNative)
+  .aggregate(
+    catsActorsJVM,
+    catsActorsJS,
+    catsActorsNative,
+    testkitJVM,
+    testkitJS,
+    testkitNative
+  )
   .settings(
     name := "cats-actors-root",
     publish / skip := true

shared/src/main/scala/com/suprnation/actor/test/TestKit.scala

@@ -107,7 +107,11 @@ trait TestKit {
         timeout,
         100.millis,
         s"expecting messages: $messages"
-      )
+      ).onError { case e =>
+        actor.messageBuffer.flatMap(q =>
+          Console[F].println(s"Expected message not found! Queue state: $q")
+        )
+      }
     } yield ()
 
   def expectMsgPF[F[+_]: Async: Console](actor: ActorRef[F, ?], timeout: FiniteDuration = 1.minute)(
@@ -125,6 +129,21 @@ trait TestKit {
       _ <- actor.messageBuffer.map { case (_, messages) => messages.collectFirst(pF) }
     } yield ()
 
+  /** Asserts that the actor receives exactly one further message, and that it is of type `T`.
+    *
+    * '''WARNING: this method can cause test flakiness if the expected message arrives quickly!!!'''
+    * Use it only when you know there is enough delay between the test calling this method and the
+    * message being added to the queue.
+    *
+    * ''Root cause (the `splitAt`):'' the buffer is snapshotted as `startQ` on entry, and the wait
+    * condition `splitAt(startQ._2.length)`s the current buffer, succeeding only when the remainder is
+    * '''exactly one''' new element. If the message is already in `startQ` (it arrived "quickly", before
+    * the snapshot), the remainder is empty and the assertion can never pass.
+    *
+    * The assertion is only reliable when delivery is guaranteed to occur strictly after this call.
+    * For a snapshot/position-independent check prefer [[expectMsgTypeCountN]], which counts messages
+    * of the type and is not sensitive to when they arrive relative to the call.
+    */
   def expectMsgType[F[+_]: Async: Console, T: ClassTag](
       actor: ActorRef[F, ?],
       timeout: FiniteDuration = 1.minute
@@ -133,15 +152,37 @@ trait TestKit {
       actor,
       timeout,
       startQ =>
-        actor.messageBuffer.map {
-          _._2.splitAt(startQ._2.length).toList match {
+        actor.messageBuffer.map { currentQ =>
+          currentQ._2.splitAt(startQ._2.length).toList match {
             case List(startQ._2, Seq(m)) if classTag[T].runtimeClass.isInstance(m) => true
             case _                                                                 => false
           }
         },
       s"of type: ${classTag[T].toString}"
     )
 
+  /** Checks if the actor has received a message of the given type
+    */
+  def expectMsgTypeSingle[F[+_]: Async: Console, T: ClassTag](
+      actor: ActorRef[F, ?],
+      timeout: FiniteDuration = 1.minute
+  ): F[Unit] =
+    expectMsgTypeCountN(actor, 1, timeout)
+
+  /** Checks if the actor has received N messages of the given type
+    */
+  def expectMsgTypeCountN[F[+_]: Async: Console, T: ClassTag](
+      actor: ActorRef[F, ?],
+      count: Int,
+      timeout: FiniteDuration = 1.minute
+  ): F[Unit] =
+    expectMsgInternal(
+      actor,
+      timeout,
+      _ => actor.messageBuffer.map(_._2.count(classTag[T].runtimeClass.isInstance) == count),
+      s"of type: ${classTag[T].toString}"
+    )
+
   def expectNoMsg[F[+_]: Async](
       actor: ActorRef[F, ?],
       timeout: FiniteDuration = 1.minute

testkit/shared/src/main/scala/com/suprnation/actor/test/ControlledTestException.scala

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2024 SuprNation
+ *
+ * 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.suprnation.actor.test
+
+import scala.concurrent.duration.FiniteDuration
+
+/** Signals a non-success outcome of a [[ControlledTestKit]] run that does not originate from the
+  * test logic itself (i.e. cancellation or exhaustion of the simulated time window).
+  */
+sealed abstract class ControlledTestException(message: String) extends RuntimeException(message)
+
+object ControlledTestException {
+
+  /** The simulated test was canceled before completing. */
+  case object Canceled extends ControlledTestException("The test was canceled.")
+
+  /** The simulated test did not finish within the allotted time window. */
+  final case class TimedOut(window: FiniteDuration)
+      extends ControlledTestException(
+        s"The test did not finish within the simulated time window ($window)."
+      )
+}

testkit/shared/src/main/scala/com/suprnation/actor/test/ControlledTestKit.scala

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2024 SuprNation
+ *
+ * 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.suprnation.actor.test
+
+import cats.effect.IO
+import cats.effect.kernel.Outcome
+import cats.effect.testkit.TestControl
+import com.suprnation.actor.ActorSystem
+
+import java.util.concurrent.TimeUnit
+import scala.concurrent.duration.FiniteDuration
+
+trait ControlledTestKit extends TestKit {
+
+  /** Creates an actor system and runs the provided test code within a simulated time environment.
+    *
+    * @tparam T             The return type of the test logic.
+    * @param testCode       The test logic to execute using the provisioned actor system.
+    * @param testTimeWindow The maximum simulated time allowed for the test execution (defaults to 90 seconds).
+    * @return               An `IO` containing the result of the test, which fails if the test errors, cancels, or times out.
+    */
+  def withActorSystemIO[T](
+      testCode: ActorSystem[IO] => IO[T],
+      testTimeWindow: FiniteDuration = FiniteDuration(90, TimeUnit.SECONDS)
+  ): IO[T] =
+    for {
+      testControl <- TestControl.execute(ActorSystem[IO](this.getClass.getName).use(testCode))
+      _ <- testControl.tickFor(testTimeWindow)
+      outcome <- testControl.results
+      result <- outcome match {
+        case Some(Outcome.Succeeded(value)) => IO.pure(value)
+        case Some(Outcome.Errored(e))       => IO.raiseError(e)
+        case Some(Outcome.Canceled())       => IO.raiseError(ControlledTestException.Canceled)
+        case None => IO.raiseError(ControlledTestException.TimedOut(testTimeWindow))
+      }
+    } yield result
+}

testkit/shared/src/test/scala/com/suprnation/actor/test/ControlledTestKitSpec.scala

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2024 SuprNation
+ *
+ * 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.suprnation.actor.test
+
+import cats.effect.testing.scalatest.AsyncIOSpec
+import cats.effect.{IO, Ref}
+import com.suprnation.actor.Actor.{Actor, Receive}
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AsyncWordSpec
+
+import scala.concurrent.duration._
+
+class ControlledTestKitSpec
+    extends AsyncWordSpec
+    with AsyncIOSpec
+    with Matchers
+    with ControlledTestKit {
+
+  "ControlledTestKit" should {
+    "advance simulated time so delayed scheduler work fires (the README example)" in
+      withActorSystemIO { system =>
+        for {
+          fired <- Ref[IO].of(false)
+          _ <- system.actorOf(
+            new Actor[IO, String] {
+              override def preStart: IO[Unit] =
+                context.system.scheduler.scheduleOnce_(5.seconds)(fired.set(true)).void
+
+              override def receive: Receive[IO, String] = { case _ => IO.unit }
+            },
+            "scheduler-actor"
+          )
+          // Written as a 6s wait, but ControlledTestKit advances simulated
+          // time, so the test still completes instantly.
+          _ <- IO.sleep(6.seconds)
+          result <- fired.get
+        } yield result shouldBe true
+      }
+
+    "not observe the scheduled work before its delay elapses" in
+      withActorSystemIO { system =>
+        for {
+          fired <- Ref[IO].of(false)
+          _ <- system.actorOf(
+            new Actor[IO, String] {
+              override def preStart: IO[Unit] =
+                context.system.scheduler.scheduleOnce_(5.seconds)(fired.set(true)).void
+
+              override def receive: Receive[IO, String] = { case _ => IO.unit }
+            },
+            "scheduler-actor"
+          )
+          _ <- IO.sleep(4.seconds)
+          result <- fired.get
+        } yield result shouldBe false
+      }
+
+    "propagate a failure raised by the test logic" in {
+      val boom = new RuntimeException("boom")
+      withActorSystemIO[Unit](_ => IO.raiseError(boom)).attempt
+        .asserting(_ shouldBe Left(boom))
+    }
+  }
+}