HTTP Mocking in Elixir with Bypass

HTTP Mocking in Elixir with Bypass

When Elixir applications call external HTTP services — payment processors, webhooks, third-party APIs — tests need to control those HTTP interactions. Bypass spins up a real HTTP server locally, intercepts requests your application makes, and lets you define exactly what responses to return. Unlike request-level stubs, Bypass tests your HTTP client configuration, URL construction, and error handling at the real TCP level.

How Bypass Works

Bypass starts a real HTTP server on a random port before each test. You configure how it responds to requests. Your application's HTTP client makes real TCP connections to this server. After the test, Bypass shuts down automatically.

This approach catches issues that stub-based mocking misses:

  • Wrong base URL configuration
  • Missing or incorrect headers
  • HTTP client timeout configuration
  • Connection pooling behavior
  • SSL/TLS configuration errors

Installation

# mix.exs
defp deps do
  [
    {:bypass, "~> 2.1", only: :test},
    {:httpoison, "~> 2.0"},  # or req, finch, tesla
  ]
end

Basic Setup

defmodule MyApp.WeatherApiTest do
  use ExUnit.Case, async: true

  setup do
    bypass = Bypass.open()
    {:ok, bypass: bypass}
  end

  test "fetches current weather", %{bypass: bypass} do
    Bypass.expect_once(bypass, "GET", "/weather/current", fn conn ->
      Plug.Conn.resp(conn, 200, Jason.encode!(%{
        temperature: 72,
        condition: "sunny",
        city: "NYC"
      }))
    end)

    result = MyApp.WeatherApi.current("NYC", base_url: "http://localhost:#{bypass.port}")

    assert {:ok, %{temperature: 72, condition: "sunny"}} = result
  end
end

Bypass.open() starts the server. Bypass.expect_once/4 defines a handler that matches one request. After the test completes, Bypass verifies the expected request was actually made — if your code never called the API, the test fails.

Configuring Your Application URL

The key to Bypass integration is making the base URL configurable. Never hardcode API URLs:

# lib/my_app/weather_api.ex
defmodule MyApp.WeatherApi do
  @base_url Application.compile_env(:my_app, :weather_api_url, "https://api.weather.com")

  def current(city, opts \\ []) do
    base_url = Keyword.get(opts, :base_url, @base_url)
    url = "#{base_url}/weather/current?city=#{city}"
    # ...
  end
end

Or use application config:

# config/test.exs
config :my_app, :weather_api_url, "http://localhost"
# In tests, override port
setup do
  bypass = Bypass.open()
  Application.put_env(:my_app, :weather_api_url, "http://localhost:#{bypass.port}")
  on_exit(fn -> Application.delete_env(:my_app, :weather_api_url) end)
  {:ok, bypass: bypass}
end

Request Verification Patterns

Verify Request Path and Method

test "calls correct endpoint", %{bypass: bypass} do
  Bypass.expect_once(bypass, "POST", "/v1/charges", fn conn ->
    {:ok, body, conn} = Plug.Conn.read_body(conn)
    payload = Jason.decode!(body)

    assert payload["amount"] == 1000
    assert payload["currency"] == "usd"

    conn
    |> Plug.Conn.put_resp_content_type("application/json")
    |> Plug.Conn.resp(200, Jason.encode!(%{id: "ch_123", paid: true}))
  end)

  {:ok, charge} = MyApp.Stripe.charge(amount: 1000, currency: "usd")
  assert charge.id == "ch_123"
end

Verify Headers

test "sends Authorization header", %{bypass: bypass} do
  Bypass.expect_once(bypass, "GET", "/api/data", fn conn ->
    auth = Plug.Conn.get_req_header(conn, "authorization")
    assert auth == ["Bearer test-api-key"]

    Plug.Conn.resp(conn, 200, Jason.encode!(%{data: []}))
  end)

  MyApp.Client.fetch_data(api_key: "test-api-key", base_url: bypass_url(bypass))
end

Verify Query Parameters

test "passes search parameters in query string", %{bypass: bypass} do
  Bypass.expect_once(bypass, "GET", "/search", fn conn ->
    params = conn.query_params

    assert params["q"] == "elixir testing"
    assert params["page"] == "2"
    assert params["per_page"] == "20"

    Plug.Conn.resp(conn, 200, Jason.encode!(%{results: [], total: 0}))
  end)

  MyApp.Search.query("elixir testing", page: 2, per_page: 20)
end

Error Scenario Testing

Bypass makes it easy to test how your application handles API failures:

HTTP Error Responses

test "returns error tuple on 401", %{bypass: bypass} do
  Bypass.expect_once(bypass, "GET", "/protected", fn conn ->
    conn
    |> Plug.Conn.put_resp_content_type("application/json")
    |> Plug.Conn.resp(401, Jason.encode!(%{error: "Unauthorized"}))
  end)

  assert {:error, :unauthorized} = MyApp.Client.get_protected()
end

test "returns error tuple on 429 rate limit", %{bypass: bypass} do
  Bypass.expect_once(bypass, "GET", "/api/data", fn conn ->
    conn
    |> Plug.Conn.put_resp_header("retry-after", "60")
    |> Plug.Conn.resp(429, Jason.encode!(%{error: "Rate limit exceeded"}))
  end)

  assert {:error, {:rate_limited, retry_after: 60}} = MyApp.Client.fetch()
end

test "handles 503 with retry", %{bypass: bypass} do
  # First call fails
  Bypass.expect_once(bypass, "GET", "/api/data", fn conn ->
    Plug.Conn.resp(conn, 503, "Service Unavailable")
  end)

  # Retry succeeds
  Bypass.expect_once(bypass, "GET", "/api/data", fn conn ->
    Plug.Conn.resp(conn, 200, Jason.encode!(%{result: "ok"}))
  end)

  assert {:ok, %{result: "ok"}} = MyApp.Client.fetch_with_retry()
end

Network Failures

test "handles connection refused", %{bypass: bypass} do
  Bypass.down(bypass)  # shut down the server

  assert {:error, :connection_refused} = MyApp.Client.fetch()
end

test "handles timeout", %{bypass: bypass} do
  Bypass.expect_once(bypass, "GET", "/slow", fn conn ->
    Process.sleep(5_000)  # exceed client timeout
    Plug.Conn.resp(conn, 200, "ok")
  end)

  assert {:error, :timeout} = MyApp.Client.fetch_slow(timeout: 100)
end

Server Errors

test "handles malformed JSON response", %{bypass: bypass} do
  Bypass.expect_once(bypass, "GET", "/api/data", fn conn ->
    Plug.Conn.resp(conn, 200, "not valid json{{{{")
  end)

  assert {:error, :invalid_response} = MyApp.Client.fetch()
end

Multiple Requests

Use Bypass.expect/3 (without _once) for endpoints called multiple times:

test "paginates through all results", %{bypass: bypass} do
  page_responses = [
    %{items: [1, 2, 3], next_page: 2},
    %{items: [4, 5, 6], next_page: 3},
    %{items: [7, 8], next_page: nil}
  ]

  call_count = :counters.new(1, [])

  Bypass.expect(bypass, "GET", "/items", fn conn ->
    idx = :counters.get(call_count, 1)
    :counters.add(call_count, 1, 1)
    response = Enum.at(page_responses, idx)
    Plug.Conn.resp(conn, 200, Jason.encode!(response))
  end)

  all_items = MyApp.Client.fetch_all()
  assert all_items == [1, 2, 3, 4, 5, 6, 7, 8]
end

Webhook Testing

Bypass is ideal for testing webhook receipt — your app receives a webhook from an external service:

test "processes Stripe webhook", %{conn: conn} do
  payload = %{
    "type" => "payment_intent.succeeded",
    "data" => %{
      "object" => %{
        "id" => "pi_123",
        "amount" => 5000,
        "currency" => "usd",
        "metadata" => %{"order_id" => "ord-456"}
      }
    }
  }

  signature = StripeSignature.compute(payload, webhook_secret: "whsec_test")

  conn
  |> put_req_header("stripe-signature", signature)
  |> post("/webhooks/stripe", Jason.encode!(payload))
  |> json_response(200)

  assert MyApp.Orders.get("ord-456").status == :paid
end

Async Tests with Bypass

Bypass works with async: true because each test gets its own server instance on a unique port:

defmodule MyApp.ClientTest do
  use ExUnit.Case, async: true

  setup do
    bypass = Bypass.open()
    {:ok, bypass: bypass}
  end

  test "test A", %{bypass: bypass} do
    # port 12345 (example)
    Bypass.expect_once(bypass, "GET", "/a", fn conn ->
      Plug.Conn.resp(conn, 200, "ok")
    end)
    # ...
  end

  test "test B", %{bypass: bypass} do
    # port 12346 — completely separate server
    Bypass.expect_once(bypass, "GET", "/b", fn conn ->
      Plug.Conn.resp(conn, 200, "ok")
    end)
    # ...
  end
end

Helper Module

Extract common patterns into a test support module:

defmodule MyApp.TestHelpers.BypassHelper do
  def bypass_url(bypass), do: "http://localhost:#{bypass.port}"

  def respond_with(bypass, method, path, status, body) do
    Bypass.expect_once(bypass, method, path, fn conn ->
      conn
      |> Plug.Conn.put_resp_content_type("application/json")
      |> Plug.Conn.resp(status, Jason.encode!(body))
    end)
  end

  def respond_with_timeout(bypass, method, path, delay_ms) do
    Bypass.expect_once(bypass, method, path, fn conn ->
      Process.sleep(delay_ms)
      Plug.Conn.resp(conn, 200, "ok")
    end)
  end
end

Beyond Bypass: Continuous API Monitoring

Bypass ensures your application handles API interactions correctly pre-deploy. But external APIs change without notice — rate limits, response format changes, authentication expiry. HelpMeTest monitors your live application's API integrations continuously, alerting you when real external services start returning errors or timeouts.

Summary

Bypass gives Elixir tests a real HTTP server to test against — catching configuration and client errors that stub-based mocking misses. Use Bypass.expect_once/4 for single-request tests and verify request headers, query parameters, and body in the handler. Test error scenarios by returning non-200 status codes or calling Bypass.down/1 for connection failures. Enable async: true freely — each test gets its own server instance.

Configure your application clients to accept a configurable base URL, and you can point them at Bypass servers in tests without modifying production code.

Read more