diff --git a/README.md b/README.md index f461a99..89e6a54 100644 --- a/README.md +++ b/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 diff --git a/build.sbt b/build.sbt index 6312c1f..eaab201 100644 --- a/build.sbt +++ b/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 diff --git a/shared/src/main/scala/com/suprnation/actor/test/TestKit.scala b/shared/src/main/scala/com/suprnation/actor/test/TestKit.scala index f81686f..a9e4f9b 100644 --- a/shared/src/main/scala/com/suprnation/actor/test/TestKit.scala +++ b/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,8 +152,8 @@ 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 } @@ -142,6 +161,28 @@ trait TestKit { 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 diff --git a/testkit/shared/src/main/scala/com/suprnation/actor/test/ControlledTestException.scala b/testkit/shared/src/main/scala/com/suprnation/actor/test/ControlledTestException.scala new file mode 100644 index 0000000..2f3eaaf --- /dev/null +++ b/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)." + ) +} diff --git a/testkit/shared/src/main/scala/com/suprnation/actor/test/ControlledTestKit.scala b/testkit/shared/src/main/scala/com/suprnation/actor/test/ControlledTestKit.scala new file mode 100644 index 0000000..32d846b --- /dev/null +++ b/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 +} diff --git a/testkit/shared/src/test/scala/com/suprnation/actor/test/ControlledTestKitSpec.scala b/testkit/shared/src/test/scala/com/suprnation/actor/test/ControlledTestKitSpec.scala new file mode 100644 index 0000000..2162ed4 --- /dev/null +++ b/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)) + } + } +}