or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

core-testing.mddeterministic-execution.mdevent-filtering.mdfsm-testing.mdindex.mdsynchronization.mdsynchronous-testing.mdtest-utilities.md
tile.json

tessl/maven-com-typesafe-akka--akka-testkit_2-12

Akka TestKit - A comprehensive testing toolkit for Actor-based systems built with the Akka framework

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
mavenpkg:maven/com.typesafe.akka/akka-testkit_2.12@2.8.x

To install, run

npx @tessl/cli install tessl/maven-com-typesafe-akka--akka-testkit_2-12@2.8.0

index.mddocs/

Akka TestKit

Akka TestKit is a comprehensive testing toolkit designed specifically for testing Actor-based systems built with the Akka framework. It provides essential utilities for testing asynchronous, concurrent actor behaviors including TestActorRef for synchronous testing of actor internals, TestKit base class with assertion helpers and timing utilities, TestProbe for creating controllable mock actors, EventFilter for testing logging and exception scenarios, and CallingThreadDispatcher for deterministic single-threaded testing.

Package Information

  • Package Name: akka-testkit
  • Package Type: maven
  • Language: Scala
  • Installation: libraryDependencies += "com.typesafe.akka" %% "akka-testkit" % "2.8.8" % Test

Core Imports

import akka.testkit._
import akka.actor.{ActorSystem, ActorRef, Props}
import scala.concurrent.duration._

For specific components:

import akka.testkit.{TestKit, TestProbe, TestActorRef, ImplicitSender}
import akka.testkit.{EventFilter, TestFSMRef, TestLatch, TestBarrier}

Basic Usage

import akka.actor.{Actor, ActorSystem, Props}
import akka.testkit.{TestKit, ImplicitSender}
import org.scalatest.BeforeAndAfterAll
import org.scalatest.wordspec.AnyWordSpecLike
import scala.concurrent.duration._

class MyActorSpec extends TestKit(ActorSystem("test-system"))
  with AnyWordSpecLike
  with ImplicitSender
  with BeforeAndAfterAll {

  override def afterAll(): Unit = {
    TestKit.shutdownActorSystem(system)
  }

  "MyActor" should {
    "respond with pong when pinged" in {
      val myActor = system.actorOf(Props[MyActor])
      myActor ! "ping"
      expectMsg(5.seconds, "pong")
    }
  }
}

class MyActor extends Actor {
  def receive = {
    case "ping" => sender() ! "pong"
  }
}

Architecture

Akka TestKit is built around several key components:

  • TestKit Base Class: Core testing functionality with message expectations, timing control, and actor lifecycle management
  • TestProbe: Lightweight actors for controlled message sending and receiving in tests
  • TestActorRef: Synchronous actor references for direct access to actor internals
  • Event Filtering: System for intercepting and asserting on log events and exceptions
  • Synchronization Utilities: Latches and barriers for coordinating concurrent test scenarios
  • Deterministic Execution: Special dispatchers and schedulers for predictable test execution

Capabilities

Core Testing Framework

The main TestKit class and related testing utilities for Actor system testing with message expectations, timing assertions, and actor lifecycle management.

abstract class TestKitBase extends TestKitUtils {
  def expectMsg[T](obj: T): T
  def expectMsg[T](max: FiniteDuration, obj: T): T
  def expectMsgType[T](implicit t: ClassTag[T]): T
  def expectNoMessage(): Unit
  def expectNoMessage(max: FiniteDuration): Unit
  def receiveN(n: Int): immutable.Seq[AnyRef]
  def receiveOne(max: Duration): AnyRef
  def within[T](max: FiniteDuration)(f: => T): T
  def awaitCond(p: => Boolean, max: Duration, interval: Duration, message: String): Unit
  def watch(ref: ActorRef): ActorRef
  def testActor: ActorRef
  def lastSender: ActorRef
}

class TestKit(_system: ActorSystem) extends TestKitBase

class TestProbe(_application: ActorSystem, name: String) extends TestKit(_application) {
  def send(actor: ActorRef, msg: Any): Unit
  def reply(msg: Any): Unit
  def ref: ActorRef
}

Core Testing Framework

Synchronous Actor Testing

TestActorRef for direct access to actor internals and synchronous testing without the actor model's asynchronous nature.

class TestActorRef[T <: Actor](_system: ActorSystem, _props: Props, _supervisor: ActorRef, name: String) 
  extends LocalActorRef {
  def underlyingActor: T
  def receive(o: Any): Unit
  def receive(o: Any, sender: ActorRef): Unit
}

object TestActorRef {
  def apply[T <: Actor: ClassTag](factory: => T)(implicit system: ActorSystem): TestActorRef[T]
  def apply[T <: Actor](props: Props)(implicit system: ActorSystem): TestActorRef[T]
}

Synchronous Actor Testing

FSM Testing

Specialized testing utilities for Finite State Machine actors with state inspection and timer management.

class TestFSMRef[S, D, T <: Actor](system: ActorSystem, props: Props, supervisor: ActorRef, name: String) 
  extends TestActorRef[T] {
  def stateName: S
  def stateData: D
  def setState(stateName: S, stateData: D): Unit
  def startTimerWithFixedDelay(name: String, msg: Any, delay: FiniteDuration): Unit
  def isTimerActive(name: String): Boolean
}

FSM Testing

Event Filtering and Logging

System for intercepting, filtering, and asserting on log events, exceptions, and dead letters during tests.

abstract class EventFilter {
  def intercept[T](code: => T)(implicit system: ActorSystem): T
  def awaitDone(max: Duration): Boolean
}

object EventFilter {
  def apply[A <: Throwable: ClassTag](
    message: String = "",
    source: String = "",
    start: String = "",
    pattern: String = "",
    occurrences: Int = 1
  ): EventFilter
  
  def error(message: String = "", occurrences: Int = 1): EventFilter
  def warning(message: String = "", occurrences: Int = 1): EventFilter
  def info(message: String = "", occurrences: Int = 1): EventFilter
  def debug(message: String = "", occurrences: Int = 1): EventFilter
}

Event Filtering and Logging

Synchronization Utilities

Utilities for coordinating concurrent test scenarios including countdown latches and cyclic barriers.

class TestLatch(count: Int = 1)(implicit system: ActorSystem) {
  def countDown(): Unit
  def isOpen: Boolean
  def open(): Unit
  def reset(): Unit
  def ready(atMost: Duration)(implicit permit: CanAwait): TestLatch
}

class TestBarrier(count: Int) {
  def await()(implicit system: ActorSystem): Unit
  def await(timeout: FiniteDuration)(implicit system: ActorSystem): Unit
  def reset(): Unit
}

Synchronization Utilities

Deterministic Execution Control

Specialized dispatchers and schedulers for predictable, deterministic test execution and timing control.

class CallingThreadDispatcher extends MessageDispatcher {
  // Executes messages on the calling thread for deterministic execution
}

class ExplicitlyTriggeredScheduler extends Scheduler {
  def timePasses(amount: FiniteDuration): Unit
  def currentTimeMs: Long
}

Deterministic Execution Control

Test Utilities and Helpers

Utility classes, actors, and functions for common testing patterns and network testing.

object TestActors {
  val echoActorProps: Props
  val blackholeProps: Props
  def forwardActorProps(ref: ActorRef): Props
}

object SocketUtil {
  def temporaryLocalPort(udp: Boolean = false): Int
  def temporaryServerAddress(address: String, udp: Boolean): InetSocketAddress
}

case class TestException(message: String) extends RuntimeException with NoStackTrace

Test Utilities and Helpers

Types

// Core testing types
trait TestKitBase {
  implicit def testKitSettings: TestKitSettings
  def testActor: ActorRef
  def system: ActorSystem
}

// AutoPilot system for automatic message handling
abstract class AutoPilot {
  def run(sender: ActorRef, msg: Any): AutoPilot
}

case object NoAutoPilot extends AutoPilot
case object KeepRunning extends AutoPilot

// Configuration settings
class TestKitSettings(config: Config) {
  val TestTimeFactor: Double
  val SingleExpectDefaultTimeout: FiniteDuration
  val ExpectNoMessageDefaultTimeout: FiniteDuration
  val TestEventFilterLeeway: FiniteDuration
  val DefaultTimeout: Timeout
}

// Implicit traits for common patterns
trait ImplicitSender extends TestKitBase {
  implicit def self: ActorRef = testActor
}

trait DefaultTimeout extends TestKitBase {
  implicit val timeout: Timeout = testKitSettings.DefaultTimeout
}

// Time dilation for test scenarios
implicit class TestDuration(val duration: FiniteDuration) extends AnyVal {
  def dilated(implicit system: ActorSystem): FiniteDuration
}