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:
- Every test starts with a clean state
- Tests don't pollute each other's data
- No explicit cleanup code needed
- 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
endThe 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
endUse 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
endTesting 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
endPure 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
endUse 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
endThe 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
endSandbox 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
endThe 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)
endTesting 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
endFor 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)
endAsync 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
endSummary
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.