Testing with Go

Use the lifecycle guide to choose from client lifetime and concurrency. The application examples verify application writes through the same fork.

Quickstart: one prepared database, isolated tests

Install the library, save this complete test file, and run it:

go get github.com/shibukawa/pgmem/pgmemtest
go test ./...

store_test.go

package store_test

import (
    "context"
    "database/sql"
    "os"
    "testing"

    "github.com/shibukawa/pgmem/pgmemtest"
)

var fx *pgmemtest.Fixture

func TestMain(m *testing.M) {
    os.Exit(pgmemtest.Run(m, pgmemtest.Options{
        Prepare: func(ctx context.Context, db *sql.DB, _ string) error {
            _, err := db.ExecContext(ctx, "CREATE TABLE orders(id int)")
            return err
        },
    }, func(f *pgmemtest.Fixture) { fx = f }))
}

func TestCreateOrder(t *testing.T) {
    t.Parallel()
    testDB := fx.For(t)
    db := testDB.DB()
    if _, err := db.Exec("INSERT INTO orders VALUES (1)"); err != nil {
        t.Fatal(err)
    }
    // testDB.DSN() and testDB.PgxPool() use this same database.
}

fx.For(t) acquires one fork. Keep the returned object when both the application and assertions need connections: DB(), PgxConn(), PgxPool() and DSN() all refer to that fork. A second For(t) deliberately acquires another database. The existing fx.DB(t) and other direct helpers still acquire a fresh fork on every call.

Choose where PostgreSQL runs

Execution	SQL connection	Package
Inside the test process	net.Pipe by default; Transport: "tcp" or "unix" for network helpers	pgmemtest
Owned child process	TCP	pgmemprocess
Owned child process	Unix domain socket	pgmemprocess with Transport: "unix"

For an external process, install the binary once:

go install github.com/shibukawa/pgmem/cmd/pgmem@latest
go get github.com/shibukawa/pgmem/pgmemprocess

Put pgmem on PATH, set PGMEM_BINARY, or pass Binary explicitly. Replace the fixture import and TestMain in the quickstart with:

import "github.com/shibukawa/pgmem/pgmemprocess"

var fx *pgmemprocess.Fixture

func TestMain(m *testing.M) {
    os.Exit(pgmemprocess.Run(m, pgmemprocess.FixtureOptions{
        Options: pgmemprocess.Options{
            Database: "app",
            Transport: "tcp", // "unix" where AF_UNIX stream sockets are supported
        },
        Prepare: func(ctx context.Context, db *sql.DB, _ string) error {
            _, err := db.ExecContext(ctx, "CREATE TABLE orders(id int)")
            return err
        },
    }, func(f *pgmemprocess.Fixture) { fx = f }))
}

The test body stays the same. Importing pgmemprocess and pgmemfixture does not compile or link the embedded engine into your test executable. The child uses stdin/stdout for lifecycle commands; SQL uses the selected network transport.

Unix mode creates private directories below /tmp on Unix, or os.TempDir() on Windows, by default. Set SocketDir to an existing short parent path when needed. Every template and fork has a distinct socket directory; reset preserves it and normal shutdown removes it. Only socket entries live on disk; database files remain in memory. AF_UNIX stream sockets work on supported Windows versions too. Datagram sockets are not used. A host without AF_UNIX support fails at listen time. A process killed without graceful cleanup may leave socket entries; remove only its abandoned private directory.

Windows stream support and its datagram limitations are documented by Microsoft and Go.

StartupTimeout, SnapshotTimeout and ForkTimeout default to 30 seconds, and ShutdownTimeout to 10 seconds. A fork wait can time out when other tests occupy the pool; close unused forks or adjust MaxForks. A shared fork also occupies one slot.

For read-only tests, fx.SharedDB() and fx.SharedPgxPool() share one prepared fork and are closed by the fixture. Acquire them during setup when possible. Set SharedReadOnly: true to guard ordinary persistent writes on these handles; the default permits writes, which affect other shared tests. Raw endpoints and isolated For(t) handles remain writable.

pgmemtest.Fixture provides three common connection forms. Use its per-test helpers only for tests that need to write or otherwise need isolation; read-only tests can share one fork as described below. It also provides PgxConn(t) when code needs one standalone pgx connection:

Handle	Method	Transport	Best for
*pgxpool.Pool	fx.PgxPool(t)	In-process, no TCP	pgx-based application code; recommended for queries
*sql.DB	fx.DB(t)	In-process, no TCP	database/sql, ORMs and tools that accept a DB handle
DSN	fx.DSN(t)	Loopback TCP	Drivers and tools that require a connection URL

The table shows the default embedded transport: the first two use Server.Dial over net.Pipe. Setting Transport to tcp or unix also switches these helpers to that network path. The DSN serves tools that need a URL. Each helper closes its handle and fork when the test ends.

Keep application datasource code unchanged

If the application normally reads DATABASE_URL or libpq’s PGHOST, PGPORT, PGUSER, and PGDATABASE, call fx.ShadowPG(t) before constructing its repository or pool. It forks the prepared snapshot, sets those variables for the test, and restores them afterward. The application still constructs its usual driver and datasource.

user_repository_test.go

func TestCreateUser(t *testing.T) {
  fx.ShadowPG(t) // fresh fork; the repository's configuration is unchanged
  repo, err := NewUserRepository() // reads DATABASE_URL as usual
  if err != nil { t.Fatal(err) }
  t.Cleanup(func() { repo.Close() })

  if err := repo.Create(t.Context(), "ada@example.com"); err != nil { t.Fatal(err) }
  if got, err := repo.FindByEmail(t.Context(), "ada@example.com"); err != nil || got == nil || got.Email != "ada@example.com" {
    t.Fatalf("user = %v, %v", got, err)
  }
}

fx.ShadowPG(t, pgmemtest.ShadowOptions{ExtraEnv: []string{"APP_DATABASE_URL"}}) also sets an application-specific DSN variable. Shared: true reuses one fork for read-only tests; writes on that fork persist into later tests. Because testing.T.Setenv changes process-wide state, these tests and their ancestors cannot call t.Parallel(). For parallel tests, use fx.DB(t) or fx.PgxPool(t) and pass the handle to the code under test.

Go has no supported way to replace a driver already registered with database/sql, or to retarget an already-open *sql.DB or pgx pool. ShadowPG therefore takes effect when the application creates its connections after the call. See API tests and E2E tests for the same setup with an HTTP server.

Database lifecycle strategy

Use a fresh server when a test needs a different schema. When tests use the same schema and seed data, initialize a baseline once, then choose how each test uses it:

• Share one fork across a read-only package.
• For tests that write, reset a reusable fork between serial tests or fork a fresh copy per test.

The prepared baseline is the parent of both shared and per-test copies; migrations and seed data do not run again on each fork.

A fresh server per test

The simplest shape: every test starts its own server and builds the schema. Nothing is shared, so it is also the slowest when migrations are long.

func TestSchemaVariant(t *testing.T) {
  s, err := pgmem.Start(t.Context(), pgmem.Options{Database: "app"})
  if err != nil {
    t.Fatal(err)
  }
  t.Cleanup(func() { s.Close() })

  db, err := sql.Open("pgx", s.DSN())
  if err != nil {
    t.Fatal(err)
  }
  t.Cleanup(func() { db.Close() })
  if _, err := db.Exec(schemaV2); err != nil {
    t.Fatal(err)
  }
  // ...
}

Prepare the baseline once

pgmemtest.Run starts a template server, runs Prepare once, snapshots the result, runs the tests and cleans up. Put migrations and seed data in Prepare; its db handle is already connected to the template. Every later fork starts from that snapshot, so tests do not repeat the setup.

package store_test

import (
  "context"
  "database/sql"
  "os"
  "testing"

  "github.com/jackc/pgx/v5"
  "github.com/jackc/pgx/v5/stdlib"
  "github.com/shibukawa/pgmem"
  "github.com/shibukawa/pgmem/pgmemtest"
)

var fx *pgmemtest.Fixture

func TestMain(m *testing.M) {
  os.Exit(pgmemtest.Run(m, pgmemtest.Options{
    Options: pgmem.Options{Database: "app"},
    Prepare: func(ctx context.Context, db *sql.DB, dsn string) error {
      if err := migrateUp(dsn); err != nil { // golang-migrate, goose, Atlas ...
        return err
      }
      return seed(ctx, db) // dbtestify, SQL, COPY ...
    },
  }, func(f *pgmemtest.Fixture) { fx = f }))
}

Prepare receives a *sql.DB already connected in-process and the DSN for tools that want a URL. The migration and seed code from the basics goes here unchanged.

Share one fork across a read-only package

If every test in a package only reads, create one fork from the prepared snapshot and share it. The fork starts with the schema and seed data from Prepare:

readOnly, err := fx.SharedDB()
if err != nil {
    panic(err) // acquire during TestMain setup
}
// Use readOnly in read-only tests; fx.Close() owns its pool and fork.
// fx.SharedPgxPool() uses the same fork.

Caution

A shared database is only safe while no test writes to it. One test’s INSERT becomes another test’s surprise, and the failure depends on test order.

Reset or fork per test for tests that write

For sequential tests that reuse a writable fork, close its connections and call Server.Reset(ctx) between tests to restore the prepared baseline. Do not reset a fork while parallel tests are using it. For isolated or parallel tests, prefer a fresh fork per test: the Fixture helpers below create it from the prepared snapshot and close it when the test ends.

database/sql

func TestCreateOrder(t *testing.T) {
  t.Parallel()
  db := fx.DB(t) // in-process connection to a fresh copy

  if _, err := db.Exec(`INSERT INTO orders (user_id, amount) VALUES (1, 500)`); err != nil {
    t.Fatal(err)
  }
}

pgx

func TestCreateOrder(t *testing.T) {
  t.Parallel()
  conn := fx.PgxConn(t)
  _, err := conn.Exec(t.Context(), `INSERT INTO orders (user_id, amount) VALUES (1, 500)`)
  if err != nil {
    t.Fatal(err)
  }
}

pgxpool

func TestService(t *testing.T) {
  t.Parallel()
  svc := NewService(fx.PgxPool(t))
  // ...
}

DSN

func TestWithURL(t *testing.T) {
  t.Parallel()
  app := NewApp(Config{DatabaseURL: fx.DSN(t)})
  // ...
}

fx.Fork(t) returns the *pgmem.Server itself when you need both DSN() and Dial.

Verify the resulting table contents

For tests of a public API, assert its observable response. For backend unit tests, checking persisted rows directly is another useful verification method: dbtestify compares the resulting table contents with an expected data set and reports a diff on mismatch.

import (
  "github.com/shibukawa/dbtestify"
  "github.com/shibukawa/dbtestify/assertdb"
)

func TestCancelOrder(t *testing.T) {
  t.Parallel()
  db := fx.DB(t)

  if err := CancelOrder(t.Context(), db, 1); err != nil {
    t.Fatal(err)
  }
  assertdb.AssertDBWithDB(t, db, dbtestify.PostgresDialect, testdata, "testdata/after_cancel.yaml", nil)
}

Parallelism and memory

Every live fork holds its own data-directory copy and buffer cache. MaxForks limits forks from one snapshot; the default is a quarter of the available memory divided by a fork’s cost (shared_buffers plus about 32 MB), 28 on a 7 GB CI runner. go test itself runs at most -parallel tests at once (default GOMAXPROCS), so raise that too when tests wait on something other than the CPU. Set it on pgmemtest.Options or SnapshotOptions to trade memory for parallelism.

pgmemtest.Options{MaxForks: 4, /* ... */}

When the cap is full, Fork(ctx) waits until a fork closes. Its context can cancel or deadline that wait; all pgmemtest fixture helpers, including direct helpers, apply ForkTimeout (30 seconds by default) on top of the test’s context. With server logging enabled, a wait longer than five seconds is logged. See limits for how this differs from snapshot and reset timeouts.

Note

Connection pools on one fork work as on a server: every connection is its own backend process. Independent tests still get the best isolation on separate forks. See limits.

Driver configuration on one fork

Configure runtime parameters or pool limits through the TestDB callbacks. The fixture retains the chosen endpoint identity and dialer, including subprocess and Unix transport.

testDB := fx.For(t)
pool := testDB.PgxPool(func(cfg *pgxpool.Config) {
    cfg.MaxConns = 4
    cfg.ConnConfig.RuntimeParams["application_name"] = "orders-test"
})
db := testDB.DB(func(cfg *pgx.ConnConfig) {
    cfg.RuntimeParams["statement_timeout"] = "1000"
})
// pool and db connect to the same fork.

Safe worker-level parallelism with ShadowPG

go test -p 4 ./... runs different packages as separate test binaries, so each package has its own environment and TestMain fixture. Keep ShadowPG tests sequential within a package. To split one package across workers, launch separate go test -run ... -count=1 processes with disjoint test selections; each process gets its own pgmem template and fork. -parallel 1 does not make t.Setenv legal inside a test that calls t.Parallel(). For true in-process parallel tests, use per-test fx.DB(t) or fx.PgxPool(t) and inject the handle instead of using ShadowPG.