feat(migrate): configure embedded Goose providers
Use provider-local SQL and Go migrations, injected slog logging, and PostgreSQL session locking. Exercise timestamp migration flows and failures with SQLite and PostgreSQL, including concurrent startup, and require PostgreSQL integration coverage in CI. Refs #7
This commit is contained in:
end committed
1 parent
46c4416ed4
commit
badeefb9a9
13 files changed
+686
No files matched your search
@@ -0,0 +1,368 @@
|
||||
package migrate_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"embed"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/stdlib"
|
||||
"github.com/pressly/goose/v3"
|
||||
_ "modernc.org/sqlite"
|
||||
|
||||
"go.auvem.com/corestack/migrate"
|
||||
)
|
||||
|
||||
// invariant: embedded fixtures are read-only; no test mutates shared migration state.
|
||||
//
|
||||
//go:embed testdata
|
||||
var fixtures embed.FS
|
||||
|
||||
func TestSQLite(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
run func(*testing.T, *sql.DB, goose.Dialect)
|
||||
}{
|
||||
{"flow", testFlow}, {"failure", testFailure},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "migrate.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := db.Close(); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
})
|
||||
test.run(t, db, goose.DialectSQLite3)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgres(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
run func(*testing.T, *sql.DB, goose.Dialect)
|
||||
}{
|
||||
{"flow", testFlow}, {"failure", testFailure},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) { test.run(t, postgresDatabase(t)(), goose.DialectPostgres) })
|
||||
}
|
||||
}
|
||||
|
||||
func testFlow(t *testing.T, db *sql.DB, dialect goose.Dialect) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
|
||||
defer cancel()
|
||||
var logs bytes.Buffer
|
||||
goMigration := goose.NewGoMigration(20261027120000,
|
||||
&goose.GoFunc{RunTx: func(ctx context.Context, tx *sql.Tx) error {
|
||||
_, err := tx.ExecContext(ctx, "INSERT INTO migration_events (version) VALUES (20261027120000)")
|
||||
return err
|
||||
}},
|
||||
&goose.GoFunc{RunTx: func(ctx context.Context, tx *sql.Tx) error {
|
||||
_, err := tx.ExecContext(ctx, "DELETE FROM migration_events WHERE version = 20261027120000")
|
||||
return err
|
||||
}})
|
||||
p, err := migrate.New(db, dialect, fixtures, "testdata/migrations", slog.New(slog.NewJSONHandler(&logs, nil)), goMigration)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
versions := []int64{7, 20261026093000, 20261027114500, 20261027120000}
|
||||
// Registering on one provider must not leak into another provider in the same process.
|
||||
other := newProvider(t, db, dialect, "testdata/migrations")
|
||||
if len(other.ListSources()) != 3 {
|
||||
t.Fatal("Go migration leaked to another provider")
|
||||
}
|
||||
assertState(t, ctx, p, 0, versions, 0)
|
||||
results, err := p.UpTo(ctx, versions[1])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, versions[:2])
|
||||
assertState(t, ctx, p, versions[1], versions, 2)
|
||||
results, err = p.Up(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, versions[2:])
|
||||
assertState(t, ctx, p, versions[3], versions, 4)
|
||||
assertEvents(t, ctx, db, versions)
|
||||
results, err = p.Up(ctx)
|
||||
if err != nil || len(results) != 0 {
|
||||
t.Fatalf("idempotent Up = %v, %v", results, err)
|
||||
}
|
||||
results, err = p.DownTo(ctx, versions[1])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, []int64{versions[3], versions[2]})
|
||||
assertState(t, ctx, p, versions[1], versions, 2)
|
||||
assertEvents(t, ctx, db, versions[:2])
|
||||
results, err = p.DownTo(ctx, versions[0])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, []int64{versions[1]})
|
||||
assertState(t, ctx, p, versions[0], versions, 1)
|
||||
assertEvents(t, ctx, db, versions[:1])
|
||||
if _, err := p.DownTo(ctx, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertState(t, ctx, p, 0, versions, 0)
|
||||
results, err = p.Up(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, versions)
|
||||
assertEvents(t, ctx, db, versions)
|
||||
if !strings.Contains(logs.String(), `"logger":"goose"`) || !strings.Contains(logs.String(), `"version":20261027114500`) {
|
||||
t.Fatalf("missing structured migration logs: %s", logs.String())
|
||||
}
|
||||
}
|
||||
|
||||
func testFailure(t *testing.T, db *sql.DB, dialect goose.Dialect) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
|
||||
defer cancel()
|
||||
p := newProvider(t, db, dialect, "testdata/failure")
|
||||
_, err := p.Up(ctx)
|
||||
var partial *goose.PartialError
|
||||
if !errors.As(err, &partial) || partial.Failed.Source.Version != 20261027114500 ||
|
||||
!strings.Contains(err.Error(), "missing_migration_table") || !strings.Contains(err.Error(), "20261027114500") {
|
||||
t.Fatalf("expected clear failed timestamp migration error, got %v", err)
|
||||
}
|
||||
assertResults(t, partial.Applied, []int64{7, 20261026093000})
|
||||
assertState(t, ctx, p, 20261026093000, []int64{7, 20261026093000, 20261027114500}, 2)
|
||||
assertEvents(t, ctx, db, []int64{7, 20261026093000})
|
||||
// A second attempt must fail at the same version without repeating committed work.
|
||||
if _, err := p.Up(ctx); err == nil {
|
||||
t.Fatal("retry unexpectedly succeeded")
|
||||
}
|
||||
assertEvents(t, ctx, db, []int64{7, 20261026093000})
|
||||
}
|
||||
|
||||
func newProvider(t *testing.T, db *sql.DB, dialect goose.Dialect, dir string, migrations ...*goose.Migration) *goose.Provider {
|
||||
t.Helper()
|
||||
p, err := migrate.New(db, dialect, fixtures, dir, slog.New(slog.NewTextHandler(io.Discard, nil)), migrations...)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func assertState(t *testing.T, ctx context.Context, p *goose.Provider, current int64, versions []int64, applied int) {
|
||||
t.Helper()
|
||||
got, err := p.GetDBVersion(ctx)
|
||||
if err != nil || got != current {
|
||||
t.Fatalf("GetDBVersion = %d, %v; want %d", got, err, current)
|
||||
}
|
||||
got, latest, err := p.GetVersions(ctx)
|
||||
if err != nil || got != current || latest != versions[len(versions)-1] {
|
||||
t.Fatalf("GetVersions = %d, %d, %v", got, latest, err)
|
||||
}
|
||||
pending, err := p.HasPending(ctx)
|
||||
if err != nil || pending != (applied < len(versions)) {
|
||||
t.Fatalf("HasPending = %v, %v", pending, err)
|
||||
}
|
||||
status, err := p.Status(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(status) != len(versions) {
|
||||
t.Fatalf("Status length = %d; want %d", len(status), len(versions))
|
||||
}
|
||||
for i, s := range status {
|
||||
want := goose.StatePending
|
||||
if i < applied {
|
||||
want = goose.StateApplied
|
||||
}
|
||||
if s.Source.Version != versions[i] || s.State != want {
|
||||
t.Fatalf("Status[%d] = %+v, version %d; want %s, %d", i, s, s.Source.Version, want, versions[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func assertResults(t *testing.T, results []*goose.MigrationResult, want []int64) {
|
||||
t.Helper()
|
||||
got := make([]int64, 0, len(results))
|
||||
for _, r := range results {
|
||||
got = append(got, r.Source.Version)
|
||||
}
|
||||
if !slices.Equal(got, want) {
|
||||
t.Fatalf("applied versions = %v; want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func assertEvents(t *testing.T, ctx context.Context, db *sql.DB, want []int64) {
|
||||
t.Helper()
|
||||
rows, err := db.QueryContext(ctx, "SELECT version FROM migration_events ORDER BY version")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var got []int64
|
||||
for rows.Next() {
|
||||
var v int64
|
||||
if err := rows.Scan(&v); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got = append(got, v)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !slices.Equal(got, want) {
|
||||
t.Fatalf("events = %v; want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// postgresDatabase owns only its random schema, never the supplied database.
|
||||
func postgresDatabase(t *testing.T) func() *sql.DB {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("MIGRATE_POSTGRES_DSN")
|
||||
if dsn == "" {
|
||||
if os.Getenv("CI") != "" {
|
||||
t.Fatal("MIGRATE_POSTGRES_DSN is required in CI")
|
||||
}
|
||||
t.Skip("set MIGRATE_POSTGRES_DSN to run Postgres integration tests")
|
||||
}
|
||||
config, err := pgx.ParseConfig(dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
admin := stdlib.OpenDB(*config)
|
||||
t.Cleanup(func() {
|
||||
if err := admin.Close(); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
})
|
||||
schema := "migrate_" + strings.ToLower(rand.Text())
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
|
||||
defer cancel()
|
||||
if _, err := admin.ExecContext(ctx, "CREATE SCHEMA "+schema); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
if _, err := admin.ExecContext(ctx, "DROP SCHEMA "+schema+" CASCADE"); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
})
|
||||
config.RuntimeParams["search_path"] = schema
|
||||
return func() *sql.DB {
|
||||
db := stdlib.OpenDB(*config)
|
||||
t.Cleanup(func() {
|
||||
if err := db.Close(); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
})
|
||||
return db
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgresConcurrentUp(t *testing.T) {
|
||||
open := postgresDatabase(t)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
|
||||
defer cancel()
|
||||
const workers = 4
|
||||
providers := make([]*goose.Provider, workers)
|
||||
for i := range providers {
|
||||
// Each provider owns a distinct pool and Go migration instance, like separate daemons.
|
||||
pause := goose.NewGoMigration(3, &goose.GoFunc{RunTx: func(ctx context.Context, tx *sql.Tx) error {
|
||||
_, err := tx.ExecContext(ctx, "SELECT pg_sleep(0.2)")
|
||||
return err
|
||||
}}, nil)
|
||||
providers[i] = newProvider(t, open(), goose.DialectPostgres, "testdata/migrations", pause)
|
||||
}
|
||||
type outcome struct {
|
||||
results []*goose.MigrationResult
|
||||
err error
|
||||
}
|
||||
start := make(chan struct{})
|
||||
done := make(chan outcome, workers)
|
||||
for _, p := range providers {
|
||||
go func() {
|
||||
select {
|
||||
case <-start:
|
||||
case <-ctx.Done():
|
||||
done <- outcome{err: ctx.Err()}
|
||||
return
|
||||
}
|
||||
results, err := p.Up(ctx)
|
||||
done <- outcome{results, err}
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
var applied []int64
|
||||
for range workers {
|
||||
result := <-done
|
||||
if result.err != nil {
|
||||
t.Error(result.err)
|
||||
}
|
||||
for _, r := range result.results {
|
||||
applied = append(applied, r.Source.Version)
|
||||
}
|
||||
}
|
||||
slices.Sort(applied)
|
||||
if want := []int64{3, 7, 20261026093000, 20261027114500}; !slices.Equal(applied, want) {
|
||||
t.Fatalf("concurrent Up applied = %v; want exactly once each: %v", applied, want)
|
||||
}
|
||||
assertEvents(t, ctx, open(), []int64{7, 20261026093000, 20261027114500})
|
||||
for _, p := range providers {
|
||||
assertState(t, ctx, p, 20261027114500, applied, len(applied))
|
||||
}
|
||||
}
|
||||
|
||||
func TestConstructor(t *testing.T) {
|
||||
db, err := sql.Open("sqlite", ":memory:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
db *sql.DB
|
||||
dialect goose.Dialect
|
||||
dir string
|
||||
logger *slog.Logger
|
||||
migrations []*goose.Migration
|
||||
contains string
|
||||
}{
|
||||
{name: "nil database", dialect: goose.DialectSQLite3, dir: "testdata/migrations", logger: logger, contains: "db must not be nil"},
|
||||
{name: "nil logger", db: db, dialect: goose.DialectSQLite3, dir: "testdata/migrations", contains: "slog logger must not be nil"},
|
||||
{name: "invalid directory", db: db, dialect: goose.DialectSQLite3, dir: "../migrations", logger: logger, contains: "directory"},
|
||||
{name: "missing directory", db: db, dialect: goose.DialectSQLite3, dir: "missing", logger: logger, contains: "migrate:"},
|
||||
{name: "invalid dialect", db: db, dialect: "unknown", dir: "testdata/migrations", logger: logger, contains: "dialect"},
|
||||
{name: "nil migration", db: db, dialect: goose.DialectSQLite3, dir: "testdata/migrations", logger: logger, migrations: []*goose.Migration{nil}, contains: "must not be nil"},
|
||||
{name: "duplicate version", db: db, dialect: goose.DialectSQLite3, dir: "testdata/migrations", logger: logger, migrations: []*goose.Migration{goose.NewGoMigration(7, nil, nil)}, contains: "7"},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
p, err := migrate.New(test.db, test.dialect, fixtures, test.dir, test.logger, test.migrations...)
|
||||
if err == nil || p != nil || !strings.Contains(err.Error(), test.contains) {
|
||||
t.Fatalf("New = %v, %v; want error containing %q", p, err, test.contains)
|
||||
}
|
||||
})
|
||||
}
|
||||
// Construction and validation are offline, even when the database is closed.
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := migrate.New(db, goose.DialectSQLite3, fixtures, "testdata/migrations", logger); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user