How to test persistent actors with ScalaTest, ZIO and Akka

Question

I have one "use case" that I would like to test. This one returns a zio.IO resulting from a call to an Actor (scala.concurrent.Future). And, I would like to use ScalaTest's AnyWordSpec.

  val actorRef = testKit.spawn(/* Real behavior */)

  "AkkaZioAndScalatest" should {

    "collaborate to write readable tests" in {
      val subject = new UseCaseWithAkkaAndZio(actorRef)
      val result =  zio.Runtime.default.unsafeRun(for {
        _ <- subject.apply("Hello")
        msg <- actorRef.askIO(GetMessage(_))
      } yield {
        msg
      })

      result should be ("Hello")
    }
  }

class UseCaseWithAkkaAndZio(actorRef: ActorRef) {
  def apply(msg:String):UIO[Unit] = {
    // askIO is an implicit method that `ask` and convert the Future to UIO
    actorRef.askIO(SetMessage(msg))
  }
}

I have turned around this test for a while without being able to have something working.

At this time, with the code sample above, I get this error:

[info] zio.FiberFailure: Fiber failed.
[info] An unchecked error was produced.
[info] java.util.concurrent.TimeoutException: Recipient[Actor[akka://UseCaseWithAkkaAndZioTest/user/$a#2047466881]] had already been terminated.
[info] at akka.actor.typed.scaladsl.AskPattern$PromiseRef.(AskPattern.scala:140)
[info] ...

Can someone help with this idea? I have to admit that I feel a bit lost here.

Thanks

---

EDIT To be complete, and as suggested by @Gaston here is the reproducible example.

Please note that it works when using a Behavior but fails when using a EventSourcedBehavior. (Thanks Gaston, I have discovered that when writing the example).

package org

import akka.actor.typed.ActorRef
import akka.cluster.sharding.typed.scaladsl.EntityRef
import akka.util.Timeout
import zio._

import scala.concurrent.duration.DurationInt

package object example {

  implicit final class EntityRefOps[-M](actorRef: EntityRef[M]) {

    private val timeout: Timeout = Timeout(5.seconds)

    def askIO[E, A](p: ActorRef[Either[E, A]] => M): ZIO[Any, E, A] = {
      IO.fromFuture(_ => actorRef.ask(p)(timeout))
        .orDie
        .flatMap(e => IO.fromEither(e))
    }

  }

}

package org.example

import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.{ActorRef, Behavior, SupervisorStrategy}
import akka.cluster.sharding.typed.scaladsl.EntityTypeKey
import akka.persistence.typed.PersistenceId
import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior, RetentionCriteria}

import scala.concurrent.duration.DurationInt

object GreeterPersistentActor {

  sealed trait Command
  case class SetMessage(message: String, replyTo: ActorRef[Either[Throwable, Unit]]) extends Command
  case class GetMessage(replyTo: ActorRef[Either[Throwable, String]]) extends Command

  val TypeKey: EntityTypeKey[Command] = EntityTypeKey[Command](getClass.getSimpleName)

  type ReplyEffect = akka.persistence.typed.scaladsl.ReplyEffect[Event, State]

  trait Event
  case class MessageSet(message: String) extends Event

  case class State(message: Option[String]) {
    def handle(cmd:Command):ReplyEffect = cmd match {
      case c: SetMessage =>
        val event = MessageSet(c.message)
        Effect.persist(event).thenReply(c.replyTo)(_ => Right(():Unit))
      case c: GetMessage =>
        message match {
          case Some(value) =>
            Effect.reply(c.replyTo)(Right(value))
          case None =>
            Effect.reply(c.replyTo)(Left(new Exception("No message set")))
        }
    }

    def apply(evt:Event):State = evt match {
      case e:MessageSet => State(Some(e.message))
    }
  }

  def apply(persistenceId: PersistenceId, message: Option[String]): Behavior[Command] = EventSourcedBehavior
    .withEnforcedReplies[Command, Event, State](
      persistenceId,
      State(message),
      (state, cmd) => state.handle(cmd),
      (state, evt) => state.apply(evt)
    )
    .withTagger(_ => Set("Sample"))
    .withRetention(
      RetentionCriteria.snapshotEvery(100, 2)
    )
    .onPersistFailure(
      SupervisorStrategy.restartWithBackoff(200.millis, 5.seconds, 0.2d)
    )

}

package org.example

import akka.cluster.sharding.typed.scaladsl.EntityRef
import zio._

class UseCaseWithAkkaAndZio(entityRefFor: String=>EntityRef[GreeterPersistentActor.Command]) {
  def apply(entityId: String, msg: String): IO[Throwable, Unit] = {
    entityRefFor(entityId).askIO(GreeterPersistentActor.SetMessage(msg, _))
  }
}

package org.example

import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
import akka.cluster.sharding.typed.scaladsl.{EntityRef, EntityTypeKey}
import akka.cluster.sharding.typed.testkit.scaladsl.TestEntityRef
import akka.persistence.typed.PersistenceId
import org.scalatest.wordspec.AnyWordSpecLike

class UseCaseWithAkkaAndZioSpec extends ScalaTestWithActorTestKit with AnyWordSpecLike {


  "AkkaZioAndScalatest" should {
    "collaborate to write readable tests" in {
      val entityId = "Sample"
      val actorRef = testKit.spawn(GreeterPersistentActor.apply(PersistenceId("Sample", entityId), None))
      val entityRefFor:String=>EntityRef[GreeterPersistentActor.Command] = TestEntityRef(
        EntityTypeKey[GreeterPersistentActor.Command]("Greeter"),
        _,
        actorRef
      )
      val subject = new UseCaseWithAkkaAndZio(entityRefFor)
      val result = zio.Runtime.default.unsafeRun(for {
        _ <- subject.apply(entityId, "Hello")
        msg <- entityRefFor(entityId).askIO(GreeterPersistentActor.GetMessage(_))
      } yield {
        println(s"Answer is $msg")
        msg
      })

      result should be("Hello")
    }
  }

}

// build.sbt
ThisBuild / scalaVersion := "2.13.11"

lazy val root = (project in file("."))
  .settings(
    name := "sample",
    libraryDependencies +=  "com.typesafe.akka" %% "akka-actor-typed" % "2.8.2",
    libraryDependencies +=  "com.typesafe.akka" %% "akka-persistence-typed" % "2.8.2",
    libraryDependencies +=  "com.typesafe.akka" %% "akka-cluster-sharding-typed" % "2.8.2",
    libraryDependencies +=  "com.typesafe.akka" %% "akka-actor-testkit-typed" % "2.8.2" % Test,
    libraryDependencies +=  "dev.zio" %% "zio" % "1.0.18",
    libraryDependencies +=  "org.scalatest" %% "scalatest" % "3.2.16" % Test,
  )

Answer 1

There is a testkit for testing persistent actors.

EventSourcedBehaviorTestKit must be mixed into the suite test

class UseCaseWithAkkaAndZioSpec extends ScalaTestWithActorTestKit(EventSourcedBehaviorTestKit.config) with AnyWordSpecLike {
  // the tests
}

EDIT:

Another question that was part of the original one was to remove the zio.Runtime.unsafeRun.

My actual solution is to create a few Matcher[ZIO[...] that allows me to write test like :

  val effect:IO[Throwable, String] = ???
  effect should completeMatching {
    case str:String => // Yes. This is useless, but for the example
  }

trait ZioMatchers {

  def completeMatching(matcher: PartialFunction[Any, Any]): Matcher[IO[Any, Any]] = (effect: IO[Any, Any]) => {
    val (complete, matches, result) = zio.Runtime.default
      .unsafeRun(
        effect
          .flatMap {
            case res if matcher.isDefinedAt(res) => ZIO.succeed((true, true, res))
            case other                           => ZIO.succeed(true, false, other)
          }
          .catchAll(err => ZIO.succeed(false, false, err))
      )
      ...
  }

  ...
}