Ecto Testing in Elixir: Sandbox, Fixtures, and Database Strategies

Ecto Testing in Elixir: Sandbox, Fixtures, and Database Strategies

Testing Elixir applications that use Ecto requires understanding one core mechanism: the SQL sandbox. Get it right, and you can run hundreds of database tests in parallel with full isolation. Get it wrong, and you have flaky tests that pass individually but fail when run together. This guide covers the sandbox, factory patterns, and the full range of Ecto testing scenarios.

Understanding the SQL Sandbox

Ecto's SQL sandbox wraps each test in a database transaction that is rolled back at the end of the test. This means:

  1. Every test starts with a clean state
  2. Tests don't pollute each other's data
  3. No explicit cleanup code needed
  4. Tests can run in parallel safely

Configure the sandbox in test/test_helper.exs:

# test/test_helper.exs
ExUnit.start()

Ecto.Adapters.SQL.Sandbox.mode(MyApp.Repo, :manual)

And in each test module that touches the database:

defmodule MyApp.AccountsTest do
  use ExUnit.Case, async: true  # async: true is safe with sandbox

  setup do
    :ok = Ecto.Adapters.SQL.Sandbox.checkout(MyApp.Repo)
  end

  test "creates a user" do
    {:ok, user} = MyApp.Accounts.create_user(%{email: "alice@example.com", name: "Alice"})
    assert user.id
    assert user.email == "alice@example.com"
  end
end

The checkout call acquires a connection for this test process. When the test ends, ExUnit automatically rolls back the transaction, returning the connection to the pool.

DataCase — The Standard Pattern

Phoenix generates a DataCase module that packages this setup:

# test/support/data_case.ex
defmodule MyApp.DataCase do
  use ExUnit.CaseTemplate

  using do
    quote do
      alias MyApp.Repo

      import Ecto
      import Ecto.Changeset
      import Ecto.Query
      import MyApp.DataCase
    end
  end

  setup tags do
    MyApp.DataCase.setup_sandbox(tags)
  end

  def setup_sandbox(tags) do
    pid = Ecto.Adapters.SQL.Sandbox.start_owner!(
      MyApp.Repo,
      shared: not tags[:async]
    )

    on_exit(fn -> Ecto.Adapters.SQL.Sandbox.stop_owner(pid) end)
  end

  def errors_on(changeset) do
    Ecto.Changeset.traverse_errors(changeset, fn {message, opts} ->
      Regex.replace(~r"%{(\w+)}", message, fn _, key ->
        opts |> Keyword.get(String.to_existing_atom(key), key) |> to_string()
      end)
    end)
  end
end

Use it in tests:

defmodule MyApp.AccountsTest do
  use MyApp.DataCase, async: true

  describe "create_user/1" do
    test "with valid data creates a user" do
      valid_attrs = %{email: "test@example.com", name: "Test User"}
      assert {:ok, user} = MyApp.Accounts.create_user(valid_attrs)
      assert user.email == "test@example.com"
    end

    test "with invalid email returns error" do
      invalid_attrs = %{email: "not-an-email", name: "Test"}
      assert {:error, changeset} = MyApp.Accounts.create_user(invalid_attrs)
      assert %{email: ["has invalid format"]} = errors_on(changeset)
    end

    test "with duplicate email returns error" do
      {:ok, _} = MyApp.Accounts.create_user(%{email: "dup@example.com", name: "First"})
      assert {:error, changeset} = MyApp.Accounts.create_user(%{email: "dup@example.com", name: "Second"})
      assert %{email: ["has already been taken"]} = errors_on(changeset)
    end
  end
end

Testing Changesets Directly

Changeset validation logic can be tested without database interaction:

defmodule MyApp.UserTest do
  use ExUnit.Case, async: true  # no DB needed, no sandbox

  alias MyApp.User

  describe "changeset/2" do
    test "valid attributes pass validation" do
      attrs = %{email: "valid@example.com", name: "Valid User", age: 25}
      changeset = User.changeset(%User{}, attrs)
      assert changeset.valid?
    end

    test "email is required" do
      changeset = User.changeset(%User{}, %{name: "No Email"})
      refute changeset.valid?
      assert {:email, {"can't be blank", _}} = hd(changeset.errors)
    end

    test "email format is validated" do
      changeset = User.changeset(%User{}, %{email: "bad", name: "Test"})
      refute changeset.valid?
      assert "has invalid format" in errors_on(changeset).email
    end

    test "age must be positive" do
      changeset = User.changeset(%User{}, %{email: "t@t.com", name: "Test", age: -1})
      refute changeset.valid?
      assert "must be greater than 0" in errors_on(changeset).age
    end
  end

  defp errors_on(changeset) do
    Ecto.Changeset.traverse_errors(changeset, fn {msg, opts} ->
      Enum.reduce(opts, msg, fn {key, value}, acc ->
        String.replace(acc, "%{#{key}}", to_string(value))
      end)
    end)
  end
end

Pure changeset tests are fast and don't need database connections — run them without DataCase.

ExMachina Factories

Hard-coded fixture data becomes unmaintainable. ExMachina provides factory functions that generate valid test data with sane defaults:

# mix.exs deps
{:ex_machina, "~> 2.7", only: :test}
# test/support/factory.ex
defmodule MyApp.Factory do
  use ExMachina.Ecto, repo: MyApp.Repo

  def user_factory do
    %MyApp.User{
      name: sequence(:name, &"User #{&1}"),
      email: sequence(:email, &"user#{&1}@example.com"),
      age: 25,
      role: :user,
      inserted_at: DateTime.utc_now(),
      updated_at: DateTime.utc_now()
    }
  end

  def admin_user_factory do
    struct!(user_factory(), role: :admin)
  end

  def post_factory do
    %MyApp.Post{
      title: sequence(:title, &"Post #{&1}"),
      body: "Some body text",
      published: false,
      author: build(:user)
    }
  end

  def published_post_factory do
    struct!(post_factory(), published: true, published_at: DateTime.utc_now())
  end
end

Use in tests:

defmodule MyApp.PostsTest do
  use MyApp.DataCase, async: true
  import MyApp.Factory

  test "lists only published posts" do
    insert(:published_post, title: "Visible Post")
    insert(:post, title: "Draft Post")  # published: false

    posts = MyApp.Posts.list_published()
    titles = Enum.map(posts, & &1.title)

    assert "Visible Post" in titles
    refute "Draft Post" in titles
  end

  test "admin can see all posts" do
    admin = insert(:admin_user)
    insert(:post, title: "Draft")
    insert(:published_post, title: "Published")

    posts = MyApp.Posts.list_all(admin)
    assert length(posts) == 2
  end
end

The sequence helper generates unique values across the test run, preventing unique constraint violations.

Testing Queries

Repository functions with complex Ecto queries need test coverage against real data:

defmodule MyApp.SearchTest do
  use MyApp.DataCase, async: true
  import MyApp.Factory

  describe "search_users/1" do
    setup do
      insert(:user, name: "Alice Smith", email: "alice@example.com")
      insert(:user, name: "Bob Jones", email: "bob@example.com")
      insert(:user, name: "Charlie Brown", email: "charlie@example.com")
      :ok
    end

    test "finds by name fragment" do
      results = MyApp.Search.search_users("alice")
      assert length(results) == 1
      assert hd(results).name == "Alice Smith"
    end

    test "is case insensitive" do
      results = MyApp.Search.search_users("ALICE")
      assert length(results) == 1
    end

    test "returns empty list for no match" do
      assert MyApp.Search.search_users("zzznomatch") == []
    end

    test "matches email domain" do
      results = MyApp.Search.search_users("example.com")
      assert length(results) == 3
    end
  end
end

Sandbox with LiveView and Channel Tests

When testing Phoenix LiveView or Channels that interact with the database, you need to allow the LiveView process to use the sandbox connection:

defmodule MyAppWeb.UserLiveTest do
  use MyAppWeb.ConnCase, async: true
  import MyApp.Factory
  import Phoenix.LiveViewTest

  test "shows user profile" do
    user = insert(:user, name: "Test User")

    # Allow the LiveView process to use sandbox
    conn = Phoenix.ConnTest.build_conn()
    {:ok, view, html} = live(conn, ~p"/users/#{user.id}")

    assert html =~ "Test User"
  end
end

The ConnCase template (generated by Phoenix) handles sandbox allowance for the connection process. For spawned processes in tests, explicitly allow them:

setup do
  :ok = Ecto.Adapters.SQL.Sandbox.checkout(MyApp.Repo)
  Ecto.Adapters.SQL.Sandbox.allow(MyApp.Repo, self(), some_pid)
end

Testing Migrations

Migration testing catches column renames, constraint changes, and data transformations that application tests miss:

defmodule MyApp.Migrations.AddUserRoleTest do
  use MyApp.DataCase

  alias Ecto.Migration.Runner

  test "migration adds role column with default" do
    # Run the specific migration
    Ecto.Migrator.run(MyApp.Repo, [
      {20240101000000, MyApp.Repo.Migrations.AddUserRole}
    ], :up, all: true)

    # Verify schema change
    result = MyApp.Repo.query!("SELECT column_default FROM information_schema.columns WHERE table_name = 'users' AND column_name = 'role'")
    assert result.rows == [["user"]]
  end
end

For data migrations specifically, seed test data before running and verify the transformation:

test "backfills username from email" do
  # Insert users without username (before migration)
  MyApp.Repo.insert_all("users", [
    %{email: "alice@example.com", inserted_at: DateTime.utc_now(), updated_at: DateTime.utc_now()}
  ])

  # Run migration
  Ecto.Migrator.up(MyApp.Repo, 20240201000000, MyApp.Repo.Migrations.BackfillUsername)

  # Verify
  users = MyApp.Repo.all(MyApp.User)
  assert Enum.all?(users, fn u -> u.username == "alice" end)
end

Async Safety Rules

The sandbox makes async: true safe for most database tests. The exceptions:

  • Schema operations (CREATE TABLE, DROP TABLE) — cannot be rolled back, must use async: false
  • Global Repo config changes — use async: false
  • Tests that start supervised processes — must explicitly allow those PIDs to use the sandbox
# Process that needs sandbox access
defmodule MyApp.BackgroundWorkerTest do
  use MyApp.DataCase, async: false  # false because spawning processes

  test "processes job from queue" do
    job = insert(:pending_job)

    # Worker will access DB — allow it
    {:ok, worker_pid} = MyApp.Worker.start_link([])
    Ecto.Adapters.SQL.Sandbox.allow(MyApp.Repo, self(), worker_pid)

    send(worker_pid, {:process, job.id})
    assert_receive {:done, ^job.id}, 5000

    updated_job = MyApp.Repo.get!(MyApp.Job, job.id)
    assert updated_job.status == :completed
  end
end

Summary

Ecto testing has one dominant concept: the SQL sandbox. Everything else — DataCase, ExMachina, ConnCase — is infrastructure to make sandbox usage convenient. Test changesets without the database (fast), test queries with real data (accurate), and use factories to build test state without brittle fixture files. The sandbox handles cleanup automatically, so the only thing you need to think about is whether a test needs async: false due to spawned processes or schema modifications.

Start now free