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