ZIO HTTP Testkit

zio-http-testkit provides testing infrastructure for ZIO HTTP applications, enabling developers to test HTTP logic without real servers, mock external dependencies, and validate WebSocket communication. Core types: TestServer, TestClient, TestChannel, HttpTestAspect.

Here's what you can do with the testkit:

import zio._
import zio.http._
import zio.test.assertTrue

// Test a server route directly
val routes = Routes(Method.GET / "users" -> Handler.text("Alice"))

// Test a client that calls external services
val testMockClient = for {
  _ <- TestClient.addRoute(Method.GET / "api" -> handler(Response.text("mock")))
  client <- ZIO.service[Client]
  response <- client(Request.get(URL.root / "api"))
} yield assertTrue(response.status == Status.Ok)

// Test WebSocket bidirectional messaging
val echoServer: WebSocketApp[Any] = Handler.webSocket { channel =>
  channel.receiveAll {
    case Read(WebSocketFrame.Text(msg)) =>
      channel.send(Read(WebSocketFrame.text(s"Echo: $msg")))
    case _ => ZIO.unit
  }
}

Motivation

Testing HTTP applications is fundamentally different from testing pure functions. Traditional approaches are painful:

Real servers in tests — Slow (seconds per test), hard to debug (network I/O), difficult to test edge cases (how do you simulate a timeout?).
Mocking HTTP libraries — You're not really testing HTTP logic, just that you call the mock correctly. If the mock diverges from real behavior, tests pass but production fails.
Tests depending on external services — Slow, flaky (services go down), hard to run locally, can't easily test error paths.

zio-http-testkit solves this by treating routes and WebSocket handlers as pure functions. You test them directly without any server infrastructure, making tests fast, deterministic, and realistic.

Installation

Add this dependency to your build.sbt:

libraryDependencies += "dev.zio" %% "zio-http-testkit" % "3.8.1" % Test

Scala versions: 2.13.x and 3.x

Overview

The testkit provides four core types working together to cover the full testing spectrum:

TestServer — Simulates an HTTP server. Configure routes, make HTTP requests via a client, verify responses. Use for integration testing multiple routes together.
TestClient — Simulates an HTTP client. Mock external service responses, verify your code makes correct HTTP calls. Use when your application depends on external APIs.
TestChannel — In-memory bidirectional message channel. Test WebSocket handlers with realistic message exchange patterns, no actual network.
HttpTestAspect — Test utility for configuring HTTP runtime modes (Dev, Prod, Preprod). Use when your handler behavior varies by deployment mode.

How They Work Together

The four core types work together in layers, each appropriate for different testing scenarios:

Layer 1: Unit Testing with Direct Route Testing (90% of tests)

When you want the fastest feedback, invoke a Handler as a pure function:

Request ──> Route.runZIO ──> Response

No server infrastructure. Each test runs in isolation, testing business logic, routing patterns, and data transformations at lightning speed. Perfect for the majority of your test suite.

Example:

zio-http-example-testing/src/test/scala/example/testing/UnitTestingIndividualHandlers.scala
package example.testing

import zio._
import zio.http._
import zio.test._

object UnitTestingIndividualHandlers extends ZIOSpecDefault {
  def spec = test("returns user data") {
    val routes = Routes(
      Method.GET / "users" / int("id") -> handler { (id: Int, _: Request) =>
        ZIO.succeed(Response.json(s"""{"id": $id, "name": "User $id"}"""))
      }
    )

    for {
      response <- routes.runZIO(Request.get(URL(Path.root / "users" / "42")))
      body <- response.body.asString
    } yield assertTrue(body.contains("User 42"))
  }
}

Layer 2: Integration Testing with External Mocking via TestClient (5% of tests)

When your handler calls external services, TestClient intercepts those calls:

Your Code ──> TestClient ──> Mocked Response
             (configured)

Your code uses the standard Client interface, unaware it's talking to a test mock. You control every response—success, failure, timeouts, anything. Test payment processors, third-party APIs, and resilience to failure without real network calls.

Example:

zio-http-example-testing/src/test/scala/example/testing/MockingExternalDependencies.scala
package example.testing

import zio._
import zio.http._
import zio.test._

object MockingExternalDependencies extends ZIOSpecDefault {
  def spec = test("calls payment processor correctly") {
    for {
      client <- ZIO.service[Client]
      // Mock the payment processor
      _ <- TestClient.addRequestResponse(
        Request.post(URL.root / "charge", Body.fromString("""{"amount": 100}""")),
        Response.json("""{"transaction_id": "tx_123", "status": "approved"}""")
      )
      // Your code calls the mocked client
      response <- client(Request.post(URL.root / "charge", Body.fromString("""{"amount": 100}""")))
      body <- response.body.asString
    } yield assertTrue(body.contains("tx_123"))
  }.provide(TestClient.layer, Scope.default)
}

Layer 3: Full Integration Testing with TestServer (5% of tests)

When you need the complete picture—multiple Routes working together, state persisting across requests—bring in TestServer:

HTTP Request ──> TestServer ──> Route Matching ──> Handler ──> Response
                (localhost / in-process)

This is where you test feature workflows. User registration followed by email verification. API calls in sequence. Routes interacting as they would in production. It's the slowest approach, but it catches integration issues that simpler tests miss.

Examples:

Multiple routes working together:

zio-http-example-testing/src/test/scala/example/testing/IntegrationTestingMultipleRoutes.scala
package example.testing

import zio._
import zio.http._
import zio.test._

object IntegrationTestingMultipleRoutes extends ZIOSpecDefault {
  def spec = test("create and retrieve user") {
    for {
      client <- ZIO.service[Client]
      port <- ZIO.serviceWithZIO[Server](_.port)
      users <- Ref.make(Map.empty[Int, String])
      nextId <- Ref.make(0)

      // Configure create and read endpoints
      _ <- TestServer.addRoutes {
        Routes(
          Method.POST / "users" -> handler { (req: Request) =>
            req.body.asString
              .flatMap { name =>
                for {
                  id <- nextId.updateAndGet(_ + 1)
                  _ <- users.update(_ + (id -> name))
                } yield Response.status(Status.Created).addHeader("X-ID", id.toString)
              }
              .catchAll { _ =>
                ZIO.succeed(Response.status(Status.BadRequest))
              }
          },
          Method.GET / "users" / int("id") -> handler { (id: Int, _: Request) =>
            users.get.map { store =>
              store.get(id) match {
                case Some(name) => Response.json(s"""{"id": $id, "name": "$name"}""")
                case None => Response.notFound
              }
            }
          }
        )
      }

      // Create a user
      createResp <- client(Request.post(URL.root.port(port) / "users", Body.fromString("Alice")))
      userId = createResp.headers.get("X-ID").flatMap(_.toIntOption).getOrElse(1)

      // Retrieve the user
      getResp <- client(Request.get(URL.root.port(port) / "users" / userId.toString))
      body <- getResp.body.asString
    } yield assertTrue(
      createResp.status == Status.Created,
      body.contains("Alice")
    )
  }.provide(TestServer.default, Client.default, Scope.default)
}

State persisting across requests:

zio-http-example-testing/src/test/scala/example/testing/TestingStateAcrossRequests.scala
package example.testing

import zio._
import zio.http._
import zio.test._

object TestingStateAcrossRequests extends ZIOSpecDefault {
  def spec = test("counter persists across requests") {
    for {
      client <- ZIO.service[Client]
      port <- ZIO.serviceWithZIO[Server](_.port)
      counter <- Ref.make(0)

      _ <- TestServer.addRoute {
        Method.GET / "increment" -> handler { (_: Request) =>
          counter.updateAndGet(_ + 1).map(n => Response.text(s"Count: $n"))
        }
      }

      // Multiple requests share the same counter
      resp1 <- client(Request.get(URL.root.port(port) / "increment"))
      body1 <- resp1.body.asString
      resp2 <- client(Request.get(URL.root.port(port) / "increment"))
      body2 <- resp2.body.asString
    } yield assertTrue(
      body1 == "Count: 1",
      body2 == "Count: 2"
    )
  }.provide(TestServer.default, Client.default, Scope.default)
}

Testing error paths:

zio-http-example-testing/src/test/scala/example/testing/ErrorHandling.scala
package example.testing

import zio._
import zio.http._
import zio.test._

object ErrorHandling extends ZIOSpecDefault {
  def spec = test("returns 401 when authorization missing") {
    for {
      client <- ZIO.service[Client]
      port <- ZIO.serviceWithZIO[Server](_.port)

      _ <- TestServer.addRoute {
        Method.GET / "protected" -> handler { (req: Request) =>
          if (req.headers.contains("Authorization"))
            ZIO.succeed(Response.ok)
          else
            ZIO.succeed(
              Response.status(Status.Unauthorized)
                .addHeader("WWW-Authenticate", "Bearer realm=\"api\"")
            )
        }
      }

      // Without auth → 401
      noAuthResp <- client(Request.get(URL.root.port(port) / "protected"))
      // With auth → 200
      withAuthResp <- client(
        Request.get(URL.root.port(port) / "protected")
          .addHeader("Authorization", "Bearer token123")
      )
    } yield assertTrue(
      noAuthResp.status == Status.Unauthorized,
      withAuthResp.status == Status.Ok
    )
  }.provide(TestServer.default, Client.default, Scope.default)
}

(source)

Layer 4: WebSocket Testing with TestChannel (Special case)

For bidirectional communication, TestChannel lets both client and server exchange messages through in-memory queues:

Client Handler ──[TestChannel]──> Server Handler
(sends messages)                    (sends responses)

Test real-time notifications, chat messages, live data streams—all without actual network I/O. Both sides run concurrently, fully controllable.

Practical example:

zio-http-example-testing/src/test/scala/example/testing/TestingWebSocketCommunication.scala
package example.testing

import zio._
import zio.http._
import zio.http.ChannelEvent.Read
import zio.test._

object TestingWebSocketCommunication extends ZIOSpecDefault {
  def spec = test("echo server echoes messages") {
    val echoServer: WebSocketApp[Any] = Handler.webSocket { channel =>
      channel.receiveAll {
        case Read(WebSocketFrame.Text(message)) =>
          channel.send(Read(WebSocketFrame.text(s"Echo: $message")))
        case _ => ZIO.unit
      }
    }

    for {
      receivedFrame <- Promise.make[Throwable, WebSocketFrame]
      testClient: WebSocketApp[Any] = Handler.webSocket { channel =>
        for {
          // Skip handshake complete event
          _ <- channel.receive
          _ <- channel.send(Read(WebSocketFrame.text("Hello")))
          response <- channel.receive
          _ <- response match {
            case Read(frame) => receivedFrame.succeed(frame)
            case _ => receivedFrame.fail(new Exception("Expected ChannelEvent.Read"))
          }
          _ <- channel.shutdown
        } yield ()
      }

      _ <- TestClient.installSocketApp(echoServer)
      _ <- ZIO.serviceWithZIO[Client](_.socket(testClient))
      frame <- receivedFrame.await
    } yield assertTrue(
      frame match {
        case WebSocketFrame.Text(msg) => msg == "Echo: Hello"
        case _ => false
      }
    )
  }.provide(TestClient.layer, Scope.default)
}

Motivation​

Installation​

Overview​

How They Work Together​

Layer 1: Unit Testing with Direct Route Testing (90% of tests)​

Layer 2: Integration Testing with External Mocking via TestClient (5% of tests)​

Layer 3: Full Integration Testing with TestServer (5% of tests)​

Layer 4: WebSocket Testing with TestChannel (Special case)​

See Also​