アプリ経由で書き込みを検証する

まずアプリが書き込み、テストが同じ DB で結果を確かめます。アプリと検証用に別々の For(t) を呼ぶと、異なるフォークを検証してしまいます。以下のサンプルは接続先を一度選び、両方へ渡します。

リポジトリのルートでバイナリをビルドし、絶対パスを設定してください。Windows では .exe の出力先を選び、利用する shell で環境変数を設定します。

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

Go HTTP

外部プロセスの TCP 接続を使います。HTTP ハンドラに渡す sql.DB と、検証用の pgx pool は一つの TestDB から取得します。HTTP サーバーを止めてから DB の接続とフォークを閉じます。

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

Node.js 24 以降が必要です。register が import 前に接続先を設定します。アプリは import 時の pool を維持し、直列の書き込みテストだけ withTestReset で戻します。次のテストは同じ 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

マーカーによる選択後に engine を作り、その engine をアプリに渡します。書き込みケースはフォーク、読み取りケースは fork=False で共有 DB を使います。engine を閉じてからフォークを解放します。

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

一つの TestDatabase を注入し、アプリと検証の両方がそこから DataSource を取得します。各 Connection は try-with-resources で閉じます。サンプルは JUnit パッケージのテストとして実行されます。

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));
        }
    }
}

各ディレクトリには依存関係の設定も含まれています。Go の replace、Node の file:、uv の editable source はこの checkout を参照します。別のプロジェクトへ移すときは公開パッケージの依存に置き換えてください。表示しているコードは実際に CI で実行するファイルから読み込んでいます。

reset が戻すのは DB の状態です。HTTP サーバーや engine、アプリのメモリ、Valkey のキー、OpenSearch の index は各 fixture が管理します。分離方式と reset の保証範囲で、クライアントの寿命と並列実行に合う方式を選べます。