feat(migrate): configure embedded Goose providers (#14)
Add embedded SQL and provider-local Go migrations with slog logging and Postgres session locking, returning Goose's Provider directly. Cover timestamp and sequential versions, failure recovery, and concurrent startup with SQLite and Postgres tests. Reuse page's Postgres CI service. Document authoring conventions and the Go template compatibility gate. Closes #7 Refs #9
This commit was merged in pull request #14.
This commit is contained in:
end committed
1 parent
c45e8abe08
commit
f0439000cb
15 files changed
+805
-56
No files matched your search
@@ -0,0 +1,457 @@
|
||||
package migrate_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"embed"
|
||||
"errors"
|
||||
"io"
|
||||
"io/fs"
|
||||
"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}, {"sql-only", testSQLOnly}, {"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}, {"sql-only", testSQLOnly}, {"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 testSQLOnly(t *testing.T, db *sql.DB, dialect goose.Dialect) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
|
||||
defer cancel()
|
||||
for _, test := range []struct {
|
||||
dir string
|
||||
versions []int64
|
||||
}{
|
||||
{"testdata/numeric", []int64{7, 10}},
|
||||
{"testdata/migrations", []int64{7, 20261026093000, 20261027114500}},
|
||||
} {
|
||||
p := newProvider(t, db, dialect, test.dir)
|
||||
assertState(t, ctx, p, 0, test.versions, 0)
|
||||
results, err := p.UpTo(ctx, test.versions[1]-1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, test.versions[:1])
|
||||
assertState(t, ctx, p, test.versions[0], test.versions, 1)
|
||||
results, err = p.Up(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertResults(t, results, test.versions[1:])
|
||||
assertState(t, ctx, p, test.versions[len(test.versions)-1], test.versions, len(test.versions))
|
||||
assertEvents(t, ctx, db, test.versions)
|
||||
results, err = p.DownTo(ctx, test.versions[1]-1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantDown := slices.Clone(test.versions[1:])
|
||||
slices.Reverse(wantDown)
|
||||
assertResults(t, results, wantDown)
|
||||
assertState(t, ctx, p, test.versions[0], test.versions, 1)
|
||||
if _, err := p.DownTo(ctx, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertState(t, ctx, p, 0, test.versions, 0)
|
||||
}
|
||||
}
|
||||
|
||||
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()
|
||||
var sources []int64
|
||||
for _, source := range p.ListSources() {
|
||||
sources = append(sources, source.Version)
|
||||
}
|
||||
if !slices.Equal(sources, versions) {
|
||||
t.Fatalf("ListSources = %v; want %v", sources, versions)
|
||||
}
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConstructorDirectory(t *testing.T) {
|
||||
db, err := sql.Open("sqlite", ":memory:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Directory validation and Go-only construction must remain offline.
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
for _, test := range []struct {
|
||||
dir string
|
||||
wantErr string
|
||||
}{
|
||||
{dir: "missing", wantErr: "directory"},
|
||||
{dir: "testdata/migrations/00007_events.sql", wantErr: "not a directory"},
|
||||
{dir: "."},
|
||||
{dir: "testdata"},
|
||||
} {
|
||||
t.Run(test.dir, func(t *testing.T) {
|
||||
p, err := migrate.New(db, goose.DialectSQLite3, fixtures, test.dir, logger, goose.NewGoMigration(10, nil, nil))
|
||||
if test.wantErr != "" {
|
||||
if err == nil || p != nil || !strings.Contains(err.Error(), test.wantErr) || !strings.Contains(err.Error(), test.dir) {
|
||||
t.Fatalf("New = %v, %v; want directory error containing %q and %q", p, err, test.wantErr, test.dir)
|
||||
}
|
||||
if test.dir == "missing" && !errors.Is(err, fs.ErrNotExist) {
|
||||
t.Fatalf("New error = %v; want fs.ErrNotExist", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if sources := p.ListSources(); len(sources) != 1 || sources[0].Version != 10 {
|
||||
t.Fatalf("ListSources = %v; want Go migration 10 only", sources)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user