417 lines
13 KiB
Go
417 lines
13 KiB
Go
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}, {"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)
|
|
}
|
|
}
|