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