Scala

Testing cats-effect IO with munit-cats-effect

HelpMeTest

HelpMeTest

— 5 min read
Testing cats-effect IO with munit-cats-effect

Testing cats-effect applications requires a test framework that understands IO. Running IO in tests naively—calling .unsafeRunSync()—works but loses meaningful error messages, leaks resources, and breaks in concurrent scenarios. munit-cats-effect provides native IO support for MUnit, making tests composable and resource-safe.

This guide covers munit-cats-effect setup, testing IO computations, Resource lifecycle testing, fiber behavior, and dependency injection patterns for the Typelevel ecosystem.

Setup

// build.sbt
libraryDependencies ++= Seq(
  "org.typelevel" %% "munit-cats-effect" % "2.0.0" % Test,
  "org.scalameta" %% "munit"             % "1.0.0" % Test
)

CatsEffectSuite

Extend CatsEffectSuite instead of munit.FunSuite. Every test can return IO[Unit] or IO[Assertion]:

import cats.effect.IO
import munit.CatsEffectSuite

class UserServiceSuite extends CatsEffectSuite {
  test("creates user successfully") {
    for {
      service <- UserService.make[IO]
      user    <- service.create("alice@example.com", "password123")
    } yield {
      assertEquals(user.email, "alice@example.com")
      assert(user.id > 0)
    }
  }

  test("returns error for duplicate email") {
    for {
      service <- UserService.make[IO]
      _       <- service.create("alice@example.com", "pass")
      result  <- service.create("alice@example.com", "pass2").attempt
    } yield {
      assert(result.isLeft)
      assert(result.left.exists(_.getMessage.contains("already exists")))
    }
  }
}

Tests returning IO are automatically executed by the framework. No .unsafeRunSync() needed.

Asserting on Effects

// Assert IO produces a value
assertIO(service.getUser(42), expectedUser)

// Assert IO raises an error
interceptIO[UserNotFoundException] {
  service.getUser(99999)
}

// Assert IO result satisfies predicate
assertIOBoolean(service.exists("alice@example.com"))

// Check with message
assertIO(
  counter.get,
  100,
  "counter should have incremented 100 times"
)

interceptIO is the IO-aware equivalent of intercept:

test("throws on invalid input") {
  interceptIO[IllegalArgumentException] {
    EmailValidator.validate[IO]("not-an-email")
  }
}

Resource Fixtures

Resource management is central to cats-effect. munit-cats-effect provides ResourceFixture for test resources that initialize before each test and clean up after:

import munit.CatsEffectSuite
import cats.effect.Resource

class DatabaseSuite extends CatsEffectSuite {
  val dbFixture: SuiteMixin { val db: TestDatabase } =
    ResourceFixture(TestDatabase.resource[IO])

  dbFixture.test("inserts and retrieves records") { db =>
    for {
      _      <- db.insert(Record(1, "test"))
      record <- db.find(1)
    } yield assertEquals(record, Some(Record(1, "test")))
  }

  dbFixture.test("returns None for missing records") { db =>
    db.find(99999).map(result => assertEquals(result, None))
  }
}

The Resource acquired before each test is guaranteed to be released — even if the test fails or throws.

Suite-Level Fixtures

For expensive resources (database connections, test containers), share across all tests in a suite:

class IntegrationSuite extends CatsEffectSuite {
  // Acquired once, shared across all tests
  val sharedDb = ResourceSuiteLocalFixture(
    "shared-db",
    PostgresTestContainer.resource[IO]
  )

  override def munitFixtures = List(sharedDb)

  test("test 1") {
    val db = sharedDb()
    // use db
  }

  test("test 2") {
    val db = sharedDb()
    // same db instance
  }
}

Testing Concurrent IO

Parallel Execution

test("runs operations in parallel") {
  val start = System.currentTimeMillis()

  val slowOp = IO.sleep(100.milliseconds) *> IO.pure(42)

  for {
    results <- List.fill(10)(slowOp).parSequence
    elapsed = System.currentTimeMillis() - start
  } yield {
    assertEquals(results, List.fill(10)(42))
    assert(elapsed < 500, s"Expected parallel execution, took ${elapsed}ms")
  }
}

Testing with TestControl

cats-effect 3 includes TestControl for deterministic concurrency testing without wall-clock time:

import cats.effect.testkit.TestControl

test("fiber scheduler is deterministic") {
  val program = for {
    ref    <- IO.ref(0)
    fiber  <- (IO.sleep(1.second) *> ref.update(_ + 1)).start
    _      <- IO.sleep(2.seconds)
    _      <- fiber.join
    result <- ref.get
  } yield result

  TestControl.executeEmbed(program).map { result =>
    assertEquals(result, 1)
  }
}

TestControl.executeEmbed runs the program with a simulated clock — time advances only when the program yields, making concurrent tests deterministic.

Racing Effects

test("race returns the faster result") {
  val fast = IO.pure("fast")
  val slow = IO.sleep(1.hour) *> IO.pure("slow")

  for {
    result <- IO.race(fast, slow)
  } yield assertEquals(result, Left("fast"))
}

Ref and State Testing

Ref is the cats-effect equivalent of AtomicReference. Test concurrent state:

test("counter is thread-safe under parallel updates") {
  for {
    counter <- IO.ref(0)
    _       <- List.fill(1000)(counter.update(_ + 1)).parSequence_
    final   <- counter.get
  } yield assertEquals(final, 1000)
}

test("Deferred resolves when completed") {
  for {
    deferred <- IO.deferred[String]
    fiber    <- deferred.get.start
    _        <- deferred.complete("result")
    result   <- fiber.joinWithNever
  } yield assertEquals(result, "result")
}

Dependency Injection Patterns

Constructor Injection

The simplest pattern — pass dependencies as constructor arguments:

class OrderServiceSuite extends CatsEffectSuite {
  val inventoryFixture = ResourceFixture(
    Resource.eval(IO.ref(Map("item-1" -> 10)))
      .map(ref => new InMemoryInventory[IO](ref))
  )

  inventoryFixture.test("completes order when in stock") { inventory =>
    val service = new OrderService[IO](inventory, new NoOpNotifier[IO])
    for {
      order <- service.placeOrder(Order("item-1", quantity = 2))
    } yield assert(order.confirmed)
  }
}

Using cats.effect.kernel.Ref for Test State

class EmailSuite extends CatsEffectSuite {
  test("sends welcome email on registration") {
    for {
      sent    <- IO.ref(List.empty[Email])
      mailer   = new TestMailer[IO](sent)
      service  = new UserService[IO](InMemoryUserRepo.empty, mailer)
      _       <- service.register("alice@example.com", "pass")
      emails  <- sent.get
    } yield {
      assertEquals(emails.size, 1)
      assertEquals(emails.head.to, "alice@example.com")
      assert(emails.head.subject.contains("Welcome"))
    }
  }
}

Testing Streams (fs2)

For services using fs2 streams:

import fs2.Stream

test("stream emits all elements") {
  val stream = Stream.range(1, 6).covary[IO]

  stream.compile.toList.map { result =>
    assertEquals(result, List(1, 2, 3, 4, 5))
  }
}

test("stream processes in batches of 10") {
  val stream = Stream.range(1, 101).covary[IO]

  stream
    .chunkN(10)
    .evalMap(batch => IO.pure(batch.toList))
    .compile
    .toList
    .map { batches =>
      assertEquals(batches.size, 10)
      assert(batches.forall(_.size == 10))
    }
}

Timeout and Retry

import scala.concurrent.duration._

test("completes within timeout") {
  val program = service.slowOperation()

  // munit timeout
  override val munitTimeout = 5.seconds

  program.map { result =>
    assertEquals(result, expectedValue)
  }
}

test("retries on transient failure") {
  import cats.syntax.all._
  import retry._

  val policy = RetryPolicies.limitRetries[IO](3)

  retryingOnAllErrors[String](
    policy = policy,
    onError = (_, _) => IO.unit
  )(flakyService.call()).map { result =>
    assertEquals(result, "success")
  }
}

Property-Based Testing with ScalaCheck

Combine munit-cats-effect with ScalaCheck for property tests over IO:

import org.scalacheck.Prop
import munit.ScalaCheckSuite

class UserPropertySuite extends CatsEffectSuite with ScalaCheckSuite {
  property("email normalization is idempotent") {
    Prop.forAll { (email: String) =>
      // Note: property tests are synchronous; use IO.unsafeRunSync for simple cases
      val normalized = EmailNormalizer.normalize(email)
      EmailNormalizer.normalize(normalized) == normalized
    }
  }
}

For IO-returning properties, use a helper:

def propIO(name: String)(io: IO[Boolean]): Unit =
  test(name) { io.map(assert(_)) }

Continuous Production Testing

munit-cats-effect ensures your IO programs work correctly in development and CI. For production assurance, HelpMeTest monitors your live application continuously — verifying that your Typelevel services remain available and correct under real load.

Summary

munit-cats-effect makes testing cats-effect applications natural: tests return IO, resources are managed safely via ResourceFixture, and concurrent behavior is testable with TestControl. The CatsEffectSuite base class handles the runtime, error reporting, and resource lifecycle — letting you focus on behavior.

Use ResourceFixture for test-scoped resources, ResourceSuiteLocalFixture for expensive shared resources, IO.ref for test state, and TestControl for deterministic concurrency testing. These patterns scale from simple unit tests to complex integration scenarios.

Read more

Start now free