ScalaTest provides comprehensive support for testing asynchronous code including Futures, eventual consistency patterns, time-based assertions, and concurrent test execution. The framework offers async test suite variants, future handling utilities, retry mechanisms, and timeout controls for robust asynchronous testing.
Async variants of all test styles that return Future[Assertion] instead of Assertion, enabling proper async test execution.
/**
* Base trait for asynchronous test suites
*/
trait AsyncTestSuite extends Suite {
/**
* Implicit execution context for Future operations
*/
implicit def executionContext: ExecutionContext
/**
* Transform assertion Future to compatible result type
*/
final def transformToOutcome(futureAssertion: Future[compatible.Assertion]): AsyncOutcome
}
/**
* Async function-based test suite
*/
abstract class AsyncFunSuite extends AsyncTestSuite with TestSuite {
/**
* Register async test returning Future[Assertion]
* @param testName the name of the test
* @param testFun async test function returning Future[Assertion]
*/
protected def test(testName: String)(testFun: => Future[compatible.Assertion]): Unit
protected def ignore(testName: String)(testFun: => Future[compatible.Assertion]): Unit
}
// Similar async variants for all test styles:
abstract class AsyncFlatSpec extends AsyncTestSuite with TestSuite
abstract class AsyncWordSpec extends AsyncTestSuite with TestSuite
abstract class AsyncFreeSpec extends AsyncTestSuite with TestSuite
abstract class AsyncFunSpec extends AsyncTestSuite with TestSuite
abstract class AsyncFeatureSpec extends AsyncTestSuite with TestSuite
abstract class AsyncPropSpec extends AsyncTestSuite with TestSuiteUsage Example:
import org.scalatest.funsuite.AsyncFunSuite
import org.scalatest.matchers.should.Matchers
import scala.concurrent.Future
import scala.concurrent.duration._
class AsyncServiceSpec extends AsyncFunSuite with Matchers {
test("async operation should complete successfully") {
val service = new AsyncUserService()
// Return Future[Assertion] - ScalaTest handles async completion
service.createUser("John", "john@example.com").map { user =>
user.name should equal("John")
user.email should equal("john@example.com")
user.id should not be empty
}
}
test("async operation should handle errors") {
val service = new AsyncUserService()
// Test async failures
recoverToSucceededIf[ValidationException] {
service.createUser("", "invalid-email")
}
}
test("multiple async operations") {
val service = new AsyncUserService()
for {
user1 <- service.createUser("Alice", "alice@example.com")
user2 <- service.createUser("Bob", "bob@example.com")
users <- service.getUsers()
} yield {
users should contain(user1)
users should contain(user2)
users should have size 2
}
}
}Test Scala Futures with patience configuration and automatic waiting.
/**
* Utilities for testing Scala Futures
*/
trait ScalaFutures extends Futures with PatienceConfiguration {
/**
* Wait for future completion and apply assertions
* @param future the Future to wait for
* @param fun assertions to apply to the completed value
*/
def whenReady[T](future: Future[T])(fun: T => Unit): Unit
/**
* Wait for future completion with custom patience config
*/
def whenReady[T](future: Future[T], config: PatienceConfig)(fun: T => Unit): Unit
/**
* Implicit conversion to enable .futureValue syntax
*/
implicit def convertScalaFuture[T](future: Future[T]): FutureValue[T]
}
/**
* Enhanced Future with testing methods
*/
final class FutureValue[T](future: Future[T]) {
/**
* Block and return the future's value (with timeout)
*/
def futureValue: T
/**
* Block and return value with custom patience
*/
def futureValue(config: PatienceConfig): T
}Usage Examples:
import org.scalatest.funsuite.AnyFunSuite
import org.scalatest.matchers.should.Matchers
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{Seconds, Span}
import scala.concurrent.Future
class FutureTestingSpec extends AnyFunSuite with Matchers with ScalaFutures {
test("testing futures with whenReady") {
val future = Future {
Thread.sleep(100)
"Hello, World!"
}
whenReady(future) { result =>
result should equal("Hello, World!")
result should startWith("Hello")
}
}
test("testing futures with futureValue") {
val future = asyncComputation()
// Block until completion and get value
future.futureValue should be > 0
future.futureValue should be <= 100
}
test("custom patience configuration") {
val slowFuture = Future {
Thread.sleep(2000)
42
}
// Custom timeout for slow operations
whenReady(slowFuture, timeout(Span(5, Seconds))) { result =>
result should equal(42)
}
}
test("testing future failures") {
val failingFuture = Future {
throw new RuntimeException("Something went wrong")
}
// Test that future fails with expected exception
whenReady(failingFuture.failed) { exception =>
exception shouldBe a[RuntimeException]
exception.getMessage should include("went wrong")
}
}
}Test systems that achieve consistency over time using retry mechanisms.
/**
* Retry assertions until they succeed or timeout
*/
trait Eventually extends PatienceConfiguration {
/**
* Retry assertion until success or timeout
* @param fun assertion block to retry
* @return successful assertion result
*/
def eventually[T](fun: => T): T
/**
* Eventually with custom patience configuration
*/
def eventually[T](config: PatienceConfig)(fun: => T): T
}Usage Examples:
import org.scalatest.funsuite.AnyFunSuite
import org.scalatest.matchers.should.Matchers
import org.scalatest.concurrent.Eventually
import org.scalatest.time.{Seconds, Millis, Span}
class EventualConsistencySpec extends AnyFunSuite with Matchers with Eventually {
test("eventually consistent cache") {
val cache = new EventuallyConsistentCache()
cache.put("key", "value")
// May not be immediately available, but should be eventually
eventually {
cache.get("key") should equal(Some("value"))
}
}
test("distributed system synchronization") {
val cluster = new DistributedCluster()
cluster.addNode("node1", "data")
// Data should replicate to all nodes eventually
eventually(timeout(Span(10, Seconds)), interval(Span(500, Millis))) {
cluster.getAllNodes().foreach { node =>
node.getData() should contain("data")
}
}
}
test("UI state changes") {
val ui = new AsyncUI()
ui.startLoading()
// UI should show loading state eventually
eventually {
ui.isLoading should be(true)
ui.getLoadingText() should equal("Loading...")
}
ui.finishLoading()
// UI should stop loading eventually
eventually {
ui.isLoading should be(false)
}
}
}Impose time constraints on test execution and operations.
/**
* Impose time limits on test operations
*/
trait TimeLimits {
/**
* Fail test if operation takes longer than specified timeout
* @param timeout maximum allowed time
* @param fun operation to time-limit
* @return result if completed within timeout
*/
def failAfter[T](timeout: Span)(fun: => T): T
/**
* Cancel test if operation takes longer than timeout
*/
def cancelAfter[T](timeout: Span)(fun: => T): T
}
/**
* Automatic time limits for all tests in suite
*/
trait TimeLimitedTests extends TimeLimits {
/**
* Default timeout applied to all tests
*/
def timeLimit: Span
/**
* Override for specific tests that need different timeouts
*/
override def withFixture(test: NoArgTest): Outcome = {
failAfter(timeLimit)(super.withFixture(test))
}
}
/**
* Async version of time-limited tests
*/
trait AsyncTimeLimitedTests extends AsyncTestSuite with TimeLimits {
def timeLimit: Span
override def withFixture(test: NoArgAsyncTest): FutureOutcome = {
val superWithFixture = super.withFixture(test)
val timedFuture = failAfter(timeLimit)(superWithFixture.toFuture)
new FutureOutcome(timedFuture)
}
}Usage Examples:
import org.scalatest.funsuite.AnyFunSuite
import org.scalatest.matchers.should.Matchers
import org.scalatest.concurrent.{TimeLimits, TimeLimitedTests}
import org.scalatest.time.{Seconds, Span}
class TimeoutSpec extends AnyFunSuite with Matchers with TimeLimits {
test("operation should complete within time limit") {
failAfter(Span(2, Seconds)) {
val result = performExpensiveCalculation()
result should be > 0
}
}
test("slow operation should timeout") {
intercept[TestFailedDueToTimeoutException] {
failAfter(Span(1, Seconds)) {
Thread.sleep(2000) // Takes longer than 1 second
"completed"
}
}
}
}
class AutoTimeLimitedSpec extends AnyFunSuite with Matchers with TimeLimitedTests {
// All tests in this suite automatically fail after 5 seconds
def timeLimit = Span(5, Seconds)
test("fast test completes normally") {
val result = quickOperation()
result should not be null
}
test("slow test gets automatic timeout") {
// This test will automatically fail after 5 seconds
Thread.sleep(6000) // Takes longer than timeLimit
}
}Configure timeout and retry behavior for async operations.
/**
* Configuration for operations that require waiting
*/
trait PatienceConfiguration {
/**
* Configuration for timeouts and retry intervals
*/
case class PatienceConfig(
timeout: Span, // Maximum time to wait
interval: Span // Time between retry attempts
)
/**
* Default patience configuration
*/
implicit def patienceConfig: PatienceConfig
/**
* Convenience methods for creating timeouts
*/
def timeout(value: Span): PatienceConfig
def interval(value: Span): PatienceConfig
}
/**
* Extended timeouts for integration testing
*/
trait IntegrationPatience extends PatienceConfiguration {
// Longer default timeouts suitable for integration tests
implicit override val patienceConfig: PatienceConfig =
PatienceConfig(timeout = Span(15, Seconds), interval = Span(150, Millis))
}Usage Examples:
import org.scalatest.time.{Seconds, Millis, Span}
import org.scalatest.concurrent.{Eventually, IntegrationPatience}
class PatienceConfigSpec extends AnyFunSuite with Matchers
with Eventually with IntegrationPatience {
test("custom patience for specific operation") {
val service = new SlowExternalService()
// Custom patience for this specific test
eventually(timeout(Span(30, Seconds)), interval(Span(1, Seconds))) {
service.isHealthy() should be(true)
}
}
test("integration test with extended patience") {
// Uses IntegrationPatience defaults (15 seconds timeout)
val database = new DatabaseConnection()
eventually {
database.isConnected() should be(true)
database.getStatus() should equal("ready")
}
}
}Coordinate multiple threads and test concurrent behavior.
/**
* Coordinate multi-threaded test scenarios
*/
trait Conductors {
/**
* Create a conductor for orchestrating concurrent test execution
*/
def conductor: Conductor
}
/**
* Conductor for multi-threaded test coordination
*/
class Conductor {
/**
* Execute function in a separate thread
* @param fun function to execute concurrently
*/
def thread[T](fun: => T): Thread
/**
* Wait for all threads to complete
*/
def whenFinished(fun: => Unit): Unit
}
/**
* Thread synchronization utilities
*/
trait Waiters {
/**
* Create a waiter for thread coordination
*/
def waiter(): Waiter
}
class Waiter {
/**
* Signal that an expected event occurred
*/
def dismiss(): Unit
/**
* Wait for expected events with timeout
*/
def await(timeout: Span = Span(150, Millis)): Unit
}Usage Examples:
import org.scalatest.funsuite.AnyFunSuite
import org.scalatest.matchers.should.Matchers
import org.scalatest.concurrent.{Conductors, Waiters}
import java.util.concurrent.atomic.AtomicInteger
class ConcurrentTestSpec extends AnyFunSuite with Matchers
with Conductors with Waiters {
test("concurrent counter increments") {
val counter = new AtomicInteger(0)
val conductor = this.conductor
conductor.thread {
for (i <- 1 to 100) {
counter.incrementAndGet()
}
}
conductor.thread {
for (i <- 1 to 100) {
counter.incrementAndGet()
}
}
conductor.whenFinished {
counter.get() should equal(200)
}
}
test("producer-consumer coordination") {
val buffer = new java.util.concurrent.ArrayBlockingQueue[String](10)
val waiter = this.waiter()
val producer = new Thread {
override def run(): Unit = {
buffer.put("item1")
buffer.put("item2")
waiter.dismiss() // Signal items are available
}
}
val consumer = new Thread {
override def run(): Unit = {
waiter.await() // Wait for items
val item1 = buffer.take()
val item2 = buffer.take()
item1 should equal("item1")
item2 should equal("item2")
}
}
producer.start()
consumer.start()
producer.join()
consumer.join()
}
}test("async service integration") {
val service = new AsyncEmailService()
for {
result <- service.sendEmail("user@example.com", "Hello", "Test message")
status <- service.getDeliveryStatus(result.messageId)
} yield {
result.success should be(true)
status should equal("delivered")
}
}test("async error handling") {
val service = new AsyncUserService()
// Test successful recovery
recoverToSucceededIf[ValidationException] {
service.createUser("", "invalid-email")
}
// Test specific error details
service.createUser("", "").failed.map { exception =>
exception shouldBe a[ValidationException]
exception.getMessage should include("name")
}
}test("event stream processing") {
val eventStream = new AsyncEventStream()
eventStream.start()
eventually {
eventStream.getProcessedCount() should be > 0
}
eventStream.stop()
eventually {
eventStream.isRunning() should be(false)
}
}// Enable parallel execution for the entire suite
class ParallelTestSuite extends AnyFunSuite with ParallelTestExecution {
// Tests in this suite run in parallel by default
test("independent test 1") {
// This can run concurrently with other tests
performIndependentOperation()
}
test("independent test 2") {
// This can also run concurrently
performAnotherIndependentOperation()
}
}