Test through the application

Let the application write, then inspect the result through the same database. Calling For(t) separately for the application and assertions would inspect different forks. These examples select one endpoint and pass it to both.

Build the binary from the repository root and export its absolute path. On Windows, choose an .exe output path and set the environment variable in your shell.

go build -o /tmp/pgmem-example ./cmd/pgmem
export PGMEM_BINARY=/tmp/pgmem-example

Go HTTP

The fixture owns a subprocess and connects over TCP. The HTTP handler receives a sql.DB; assertions use a pgx pool from the same TestDB. Stop the HTTP server before closing the database clients and fork.

go -C examples/go-http test -v

examples/go-http/app.go

package orders

import (
  "database/sql"
  "encoding/json"
  "net/http"
)

// NewHandler receives its database before it creates any clients.
func NewHandler(db *sql.DB) http.Handler {
  mux := http.NewServeMux()
  mux.HandleFunc("POST /orders", func(w http.ResponseWriter, r *http.Request) {
    var order struct {
      ID int `json:"id"`
    }
    if err := json.NewDecoder(r.Body).Decode(&order); err != nil || order.ID <= 0 {
      http.Error(w, "invalid order", http.StatusBadRequest)
      return
    }
    if _, err := db.ExecContext(r.Context(), "INSERT INTO orders(id) VALUES ($1)", order.ID); err != nil {
      http.Error(w, "cannot create order", http.StatusInternalServerError)
      return
    }
    w.WriteHeader(http.StatusCreated)
  })
  return mux
}

examples/go-http/app_test.go

package orders

import (
  "context"
  "database/sql"
  "net/http"
  "net/http/httptest"
  "os"
  "strings"
  "testing"

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

var fixture *pgmemprocess.Fixture

func TestMain(m *testing.M) {
  os.Exit(pgmemprocess.Run(m, pgmemprocess.FixtureOptions{
    Options: pgmemprocess.Options{Transport: "tcp", Database: "orders"},
    Prepare: func(ctx context.Context, db *sql.DB, _ string) error {
      _, err := db.ExecContext(ctx, "CREATE TABLE orders(id int PRIMARY KEY)")
      return err
    },
  }, func(f *pgmemprocess.Fixture) { fixture = f }))
}

func TestCreateOrder(t *testing.T) {
  t.Parallel()
  database := fixture.For(t)
  server := httptest.NewServer(NewHandler(database.DB()))
  t.Cleanup(server.Close) // HTTP server closes before the database handles.
  response, err := http.Post(server.URL+"/orders", "application/json", strings.NewReader(`{"id":42}`))
  if err != nil {
    t.Fatal(err)
  }
  response.Body.Close()
  if response.StatusCode != http.StatusCreated {
    t.Fatalf("status: %d", response.StatusCode)
  }
  var id int
  if err := database.PgxPool().QueryRow(t.Context(), "SELECT id FROM orders").Scan(&id); err != nil || id != 42 {
    t.Fatalf("application write missing from same fork: %d %v", id, err)
  }
}

func TestAnotherTestStartsEmpty(t *testing.T) {
  t.Parallel()
  var count int
  if err := fixture.For(t).DB().QueryRow("SELECT count(*) FROM orders").Scan(&count); err != nil || count != 0 {
    t.Fatalf("another test leaked data: %d %v", count, err)
  }
}

Node HTTP

Use Node.js 24 or newer. Registration sets the endpoint before application imports. The application keeps its import-time pool; withTestReset restores the database after the serial writing case. The next case reads the baseline through that same pool.

npm --prefix examples/node-http ci --omit=optional --ignore-scripts
npm --prefix examples/node-http test

examples/node-http/app.mjs

import { createServer } from 'node:http';
import pg from 'pg';

// register must inject DATABASE_URL before this module is imported.
export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
export function createApp() {
  return createServer(async (request, response) => {
    if (request.method !== 'POST' || request.url !== '/orders') {
      response.writeHead(404).end();
      return;
    }
    try {
      let body = '';
      for await (const chunk of request) body += chunk;
      const { id } = JSON.parse(body);
      if (!Number.isSafeInteger(id) || id <= 0) { response.writeHead(400).end(); return; }
      await pool.query('INSERT INTO orders(id) VALUES ($1)', [id]);
      response.writeHead(201).end();
    } catch { response.writeHead(500).end(); }
  });
}

examples/node-http/global-setup.mjs

import pg from 'pg';
import { PgmemServer } from '@pgmem/core';
import { installServiceEnv } from '@pgmem/core/environment';

let server;
let release;
export async function globalSetup() {
  server = await PgmemServer.start({
    async prepare({ url }) {
      const pool = new pg.Pool({ connectionString: url });
      try { await pool.query('CREATE TABLE orders(id int PRIMARY KEY)'); }
      finally { await pool.end(); }
    },
  });
  try { release = installServiceEnv('pgmem:control', server.env()); }
  catch (error) {
    try { await server.close(); }
    catch (cleanup) { throw new AggregateError([error, cleanup], 'setup and cleanup failed'); }
    throw error;
  }
}
export async function globalTeardown() {
  const failures = [];
  try { release?.(); } catch (error) { failures.push(error); }
  try { await server?.close(); } catch (error) { failures.push(error); }
  if (failures.length) throw new AggregateError(failures, 'global cleanup failed');
}

examples/node-http/app.test.mjs

import { test, after, before } from 'node:test';
import assert from 'node:assert/strict';
import { once } from 'node:events';
import pg from 'pg';
import { withTestReset } from '@pgmem/core';
import { createApp, pool } from './app.mjs';

const app = createApp();
const assertions = new pg.Pool({ connectionString: process.env.DATABASE_URL });
before(async () => { app.listen(0, '127.0.0.1'); await once(app, 'listening'); });
after(async () => {
  const failures = [];
  try { await new Promise((resolve, reject) => app.close((error) => error ? reject(error) : resolve())); }
  catch (error) { failures.push(error); }
  const results = await Promise.allSettled([pool.end(), assertions.end()]);
  failures.push(...results.filter((r) => r.status === 'rejected').map((r) => r.reason));
  if (failures.length) throw new AggregateError(failures, 'application cleanup failed');
});

test('the application and assertions use the same fork', withTestReset(async () => {
  const response = await fetch(`http://127.0.0.1:${app.address().port}/orders`, {
    method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ id: 42 }),
  });
  await response.arrayBuffer();
  assert.equal(response.status, 201);
  assert.deepEqual((await assertions.query('SELECT id FROM orders')).rows, [{ id: 42 }]);
}));

test('the same application client sees the restored baseline', async () => {
  assert.equal((await pool.query('SELECT count(*)::int AS n FROM orders')).rows[0].n, 0);
});

Python SQLAlchemy

Create the engine after marker selection and pass it into the application. The writing case gets a fork; the reading case selects a shared database with fork=False. Dispose the engine before releasing its fork.

uv run --project examples/python-sqlalchemy pytest examples/python-sqlalchemy -q

examples/python-sqlalchemy/app.py

from sqlalchemy import Column, Integer, MetaData, Table, insert

metadata = MetaData()
orders = Table("orders", metadata, Column("id", Integer, primary_key=True))


class OrderApplication:
    def __init__(self, engine):
        self.engine = engine

    def create_order(self, order_id):
        with self.engine.begin() as connection:
            connection.execute(insert(orders).values(id=order_id))

examples/python-sqlalchemy/conftest.py

import pytest
from sqlalchemy import create_engine
from app import metadata, OrderApplication


def engine_for(dsn):
    return create_engine(dsn.replace("postgres://", "postgresql+psycopg://", 1))


@pytest.fixture(scope="session")
def pgmem_prepare():
    def prepare(server):
        engine = engine_for(server.dsn)
        try:
            metadata.create_all(engine)
        finally:
            engine.dispose()
    return prepare


@pytest.fixture
def engine(pgmem_test_dsn):
    value = engine_for(pgmem_test_dsn)
    try:
        yield value
    finally:
        value.dispose()


@pytest.fixture
def app(engine):
    return OrderApplication(engine)

examples/python-sqlalchemy/test_app.py

import pytest
from sqlalchemy import select, func
from app import orders


def test_application_write_is_visible_to_assertions(app, engine):
    app.create_order(42)
    with engine.connect() as connection:
        assert connection.execute(select(orders.c.id)).scalar_one() == 42


@pytest.mark.pgmem(fork=False)
def test_read_the_prepared_baseline(app, engine):
    with engine.connect() as connection:
        assert connection.execute(select(func.count()).select_from(orders)).scalar_one() == 0

Java DataSource

Inject one TestDatabase and obtain both application and assertion DataSources from it. Close each connection with try-with-resources. This example runs as part of the JUnit package tests.

cd packages/java
./gradlew :pgmem-junit5:test --tests '*OrderApplicationTest'

packages/java/pgmem-junit5/src/test/java/io/github/shibukawa/pgmem/junit5/examples/OrderApplicationTest.java

package io.github.shibukawa.pgmem.junit5.examples;

import static org.junit.jupiter.api.Assertions.assertEquals;
import java.sql.SQLException;
import javax.sql.DataSource;
import io.github.shibukawa.pgmem.junit5.PgmemExtension;
import io.github.shibukawa.pgmem.junit5.TestDatabase;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;

class OrderApplicationTest {
    @RegisterExtension static PgmemExtension pg = PgmemExtension.builder()
            .prepare(server -> {
                try (var connection = server.dataSource().getConnection(); var statement = connection.createStatement()) {
                    statement.execute("CREATE TABLE orders(id int PRIMARY KEY)");
                } catch (SQLException error) { throw new RuntimeException(error); }
            }).build();

    static final class OrderApplication {
        private final DataSource database;
        OrderApplication(DataSource database) { this.database = database; }
        void createOrder(int id) throws SQLException {
            try (var connection = database.getConnection(); var statement = connection.prepareStatement("INSERT INTO orders(id) VALUES (?)")) {
                statement.setInt(1, id);
                statement.executeUpdate();
            }
        }
    }

    @Test void applicationAndAssertionsUseTheSameFork(TestDatabase database) throws Exception {
        var application = new OrderApplication(database.dataSource());
        application.createOrder(42);
        try (var connection = database.dataSource().getConnection();
             var statement = connection.createStatement(); var rows = statement.executeQuery("SELECT id FROM orders")) {
            rows.next();
            assertEquals(42, rows.getInt(1));
        }
    }
}

Dependency manifests are included in each example directory. The Go replace, Node file: dependency and uv editable source point into this checkout. Replace them with published package dependencies when copying an example into another project. The code shown here is loaded from the files that CI executes.

A reset restores database state. Each fixture owns its HTTP server or engine, application memory, Valkey keys and OpenSearch indexes. See isolation choices and reset guarantees for a strategy that matches client lifetime and concurrency.