Repository: ariga/atlas
Pull request #2993 (2024-07-22 11:32:18)

cmd/atlas/cmdlog: move migrate status reporting to cmdlog


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 54e6beda..2ebf58b9 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -1226,7 +1226,7 @@ func migrateStatusRun(cmd *cobra.Command, _ []string, flags migrateStatusFlags)
 	if err := checkRevisionSchemaClarity(cmd, client, flags.revisionSchema); err != nil {
 		return err
 	}
-	report, err := (&cmdmigrate.StatusReporter{
+	report, err := (&cmdlog.StatusReporter{
 		Client: client,
 		Dir:    dir,
 		DirURL: dirURL,
diff --git a/cmd/atlas/internal/cmdapi/schema.go b/cmd/atlas/internal/cmdapi/schema.go
index 0a2c9624..51c1ba71 100644
--- a/cmd/atlas/internal/cmdapi/schema.go
+++ b/cmd/atlas/internal/cmdapi/schema.go
@@ -23,6 +23,7 @@ import (
 	"ariga.io/atlas/sql/sqlclient"
 
 	"github.com/1lann/promptui"
+	"github.com/chzyer/readline"
 	"github.com/hashicorp/hcl/v2/hclwrite"
 	"github.com/spf13/cobra"
 )
@@ -637,11 +638,7 @@ func setSchemaEnvFlags(cmd *cobra.Command, env *Env) error {
 	if err != nil {
 		return err
 	}
-	for i, s := range srcs {
-		if !isURL(s) {
-			srcs[i] = "file://" + s
-		}
-	}
+	srcs = fixFileURLs(srcs)
 	if err := maySetFlag(cmd, flagFile, strings.Join(srcs, ",")); err != nil {
 		return err
 	}
@@ -759,7 +756,7 @@ func cmdPrompt(cmd *cobra.Command) *promptui.Select {
 		Label:    "Are you sure?",
 		HideHelp: true,
 		Stdin:    io.NopCloser(cmd.InOrStdin()),
-		Stdout:   nopCloser{cmd.OutOrStdout()},
+		Stdout:   nopBellCloser{cmd.OutOrStdout()},
 	}
 }
 
@@ -774,9 +771,16 @@ func promptUser(cmd *cobra.Command) bool {
 	return result == answerApply
 }
 
-type nopCloser struct{ io.Writer }
+type nopBellCloser struct{ io.Writer }
+
+func (n nopBellCloser) Write(p []byte) (int, error) {
+	if len(p) == 1 && p[0] == readline.CharBell {
+		return 0, nil // Skip bell noise.
+	}
+	return n.Writer.Write(p)
+}
 
-func (nopCloser) Close() error { return nil }
+func (nopBellCloser) Close() error { return nil }
 
 func tasks(path string) ([]fmttask, error) {
 	var tasks []fmttask
@@ -832,3 +836,14 @@ func fmtFile(task fmttask) (bool, error) {
 	}
 	return false, nil
 }
+
+// fixFileURLs converts all file paths to a URL format, if not already.
+// For example, "schema.hcl" to "file://schema.hcl".
+func fixFileURLs(src []string) []string {
+	for i, s := range src {
+		if !isURL(s) {
+			src[i] = "file://" + s
+		}
+	}
+	return src
+}
diff --git a/cmd/atlas/internal/cmdlog/cmdlog.go b/cmd/atlas/internal/cmdlog/cmdlog.go
index 18900864..430dcb83 100644
--- a/cmd/atlas/internal/cmdlog/cmdlog.go
+++ b/cmd/atlas/internal/cmdlog/cmdlog.go
@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/url"
 	"slices"
@@ -16,6 +17,8 @@ import (
 	"text/template"
 	"time"
 
+	cmdmigrate "ariga.io/atlas/cmd/atlas/internal/migrate"
+	"ariga.io/atlas/cmd/atlas/internal/migrate/ent/revision"
 	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
 	"ariga.io/atlas/sql/sqlclient"
@@ -186,6 +189,111 @@ func (r *MigrateStatus) FromCheckpoint() bool {
 	return ok && ck.IsCheckpoint()
 }
 
+// StatusReporter is used to gather information about migration status.
+type StatusReporter struct {
+	// Client configures the connection to the database to file a MigrateStatus for.
+	Client *sqlclient.Client
+	// DirURL of the migration directory.
+	DirURL *url.URL
+	// Dir is used for scanning and validating the migration directory.
+	Dir migrate.Dir
+	// Schema name the revision table resides in.
+	Schema string
+}
+
+// Report creates and writes a MigrateStatus.
+func (r *StatusReporter) Report(ctx context.Context) (*MigrateStatus, error) {
+	rep := &MigrateStatus{Env: NewEnv(r.Client, r.DirURL)}
+	// Check if there already is a revision table in the defined schema.
+	// Inspect schema and check if the table does already exist.
+	sch, err := r.Client.InspectSchema(ctx, r.Schema, &schema.InspectOptions{Tables: []string{revision.Table}})
+	if err != nil && !schema.IsNotExistError(err) {
+		return nil, err
+	}
+	if schema.IsNotExistError(err) || func() bool { _, ok := sch.Table(revision.Table); return !ok }() {
+		// Either schema or table does not exist.
+		if rep.Available, err = migrate.FilesFromLastCheckpoint(r.Dir); err != nil {
+			return nil, err
+		}
+		rep.Pending = rep.Available
+	} else {
+		// Both exist, fetch their data.
+		rrw, err := cmdmigrate.RevisionsForClient(ctx, r.Client, r.Schema)
+		if err != nil {
+			return nil, err
+		}
+		if err := rrw.Migrate(ctx); err != nil {
+			return nil, err
+		}
+		ex, err := migrate.NewExecutor(r.Client.Driver, r.Dir, rrw)
+		if err != nil {
+			return nil, err
+		}
+		rep.Applied, err = rrw.ReadRevisions(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if rep.Pending, err = ex.Pending(ctx); err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
+			if err1 := (*migrate.HistoryNonLinearError)(nil); errors.As(err, &err1) {
+				rep.Error = err1.Error()
+				rep.Status = statusPending
+				rep.Pending = err1.Pending
+				rep.OutOfOrder = err1.OutOfOrder
+				// Non-linear error means at least one file was applied.
+				rep.Current = rep.Applied[len(rep.Applied)-1].Version
+				return rep, nil
+			}
+			return nil, err
+		}
+		// If no files were applied, all pending files are
+		// available. The first one might be a checkpoint.
+		if len(rep.Applied) == 0 {
+			rep.Available = rep.Pending
+		} else if rep.Available, err = r.Dir.Files(); err != nil {
+			return nil, err
+		}
+	}
+	switch len(rep.Pending) {
+	case len(rep.Available):
+		rep.Current = "No migration applied yet"
+	default:
+		rep.Current = rep.Applied[len(rep.Applied)-1].Version
+	}
+	if len(rep.Pending) == 0 {
+		rep.Status = statusOK
+		rep.Next = "Already at latest version"
+	} else {
+		rep.Status = statusPending
+		rep.Next = rep.Pending[0].Version()
+	}
+	// If the last one is partially applied (and not manually resolved).
+	if len(rep.Applied) != 0 {
+		last := rep.Applied[len(rep.Applied)-1]
+		if !last.Type.Has(migrate.RevisionTypeResolved) && last.Applied < last.Total {
+			rep.SQL = strings.ReplaceAll(last.ErrorStmt, "\n", " ")
+			rep.Error = strings.ReplaceAll(last.Error, "\n", " ")
+			rep.Count = last.Applied
+			idx := migrate.FilesLastIndex(rep.Available, func(f migrate.File) bool {
+				return f.Version() == last.Version
+			})
+			if idx == -1 {
+				return nil, fmt.Errorf("migration file with version %q not found", last.Version)
+			}
+			stmts, err := migrate.FileStmts(r.Client.Driver, rep.Available[idx])
+			if err != nil {
+				return nil, err
+			}
+			rep.Total = len(stmts)
+		}
+	}
+	return rep, nil
+}
+
+const (
+	statusOK      = "OK"
+	statusPending = "PENDING"
+)
+
 // MigrateSetTemplate holds the default template of the 'migrate set' command.
 var MigrateSetTemplate = template.Must(template.New("set").
 	Funcs(ColorTemplateFuncs).Parse(`
@@ -812,38 +920,7 @@ func (s *SchemaInspect) MarshalSQL(indent ...string) (string, error) {
 }
 
 func sqlInspect(report *SchemaInspect, indent ...string) (string, error) {
-	var changes schema.Changes
-	for _, o := range report.Realm.Objects {
-		changes = append(changes, &schema.AddObject{O: o})
-	}
-	for _, s := range report.Realm.Schemas {
-		// Generate commands for creating the schemas on realm-mode.
-		if report.client.URL.Schema == "" {
-			changes = append(changes, &schema.AddSchema{S: s})
-		}
-		for _, o := range s.Objects {
-			changes = append(changes, &schema.AddObject{O: o})
-		}
-		for _, t := range s.Tables {
-			changes = append(changes, &schema.AddTable{T: t})
-			for _, r := range t.Triggers {
-				changes = append(changes, &schema.AddTrigger{T: r})
-			}
-		}
-		for _, v := range s.Views {
-			changes = append(changes, &schema.AddView{V: v})
-			for _, r := range v.Triggers {
-				changes = append(changes, &schema.AddTrigger{T: r})
-			}
-		}
-		for _, f := range s.Funcs {
-			changes = append(changes, &schema.AddFunc{F: f})
-		}
-		for _, p := range s.Procs {
-			changes = append(changes, &schema.AddProc{P: p})
-		}
-	}
-	return fmtPlan(report.ctx, report.client, changes, indent)
+	return fmtPlan(report.ctx, report.client, cmdmigrate.ChangesToRealm(report.client, report.Realm), indent)
 }
 
 // SchemaDiff contains a summary of the 'schema diff' command.
@@ -900,7 +977,7 @@ func fmtPlan(ctx context.Context, client *sqlclient.Client, changes schema.Chang
 	}
 	plan, err := client.PlanChanges(ctx, "plan", changes, func(o *migrate.PlanOptions) {
 		o.Mode = migrate.PlanModeDump
-		// Disable tables qualifier in schema-mode.
+		// Disable object qualifier in schema-mode.
 		if client.URL.Schema != "" {
 			o.SchemaQualifier = new(string)
 		}
diff --git a/cmd/atlas/internal/migrate/migrate.go b/cmd/atlas/internal/migrate/migrate.go
index 5844ab95..4f864fb8 100644
--- a/cmd/atlas/internal/migrate/migrate.go
+++ b/cmd/atlas/internal/migrate/migrate.go
@@ -378,3 +378,39 @@ func DirURL(ctx context.Context, u *url.URL, create bool) (migrate.Dir, error) {
 	}
 	return d, err
 }
+
+// ChangesToRealm returns the schema changes for creating the given Realm.
+func ChangesToRealm(c *sqlclient.Client, r *schema.Realm) schema.Changes {
+	var changes schema.Changes
+	for _, o := range r.Objects {
+		changes = append(changes, &schema.AddObject{O: o})
+	}
+	for _, s := range r.Schemas {
+		// Generate commands for creating the schemas on realm-mode.
+		if c.URL.Schema == "" {
+			changes = append(changes, &schema.AddSchema{S: s})
+		}
+		for _, o := range s.Objects {
+			changes = append(changes, &schema.AddObject{O: o})
+		}
+		for _, t := range s.Tables {
+			changes = append(changes, &schema.AddTable{T: t})
+			for _, r := range t.Triggers {
+				changes = append(changes, &schema.AddTrigger{T: r})
+			}
+		}
+		for _, v := range s.Views {
+			changes = append(changes, &schema.AddView{V: v})
+			for _, r := range v.Triggers {
+				changes = append(changes, &schema.AddTrigger{T: r})
+			}
+		}
+		for _, f := range s.Funcs {
+			changes = append(changes, &schema.AddFunc{F: f})
+		}
+		for _, p := range s.Procs {
+			changes = append(changes, &schema.AddProc{P: p})
+		}
+	}
+	return changes
+}
diff --git a/cmd/atlas/internal/migrate/report.go b/cmd/atlas/internal/migrate/report.go
deleted file mode 100644
index f2b22685..00000000
--- a/cmd/atlas/internal/migrate/report.go
+++ /dev/null
@@ -1,124 +0,0 @@
-// Copyright 2021-present The Atlas Authors. All rights reserved.
-// This source code is licensed under the Apache 2.0 license found
-// in the LICENSE file in the root directory of this source tree.
-
-package migrate
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"net/url"
-	"strings"
-
-	"ariga.io/atlas/cmd/atlas/internal/cmdlog"
-	"ariga.io/atlas/cmd/atlas/internal/migrate/ent/revision"
-	"ariga.io/atlas/sql/migrate"
-	"ariga.io/atlas/sql/schema"
-	"ariga.io/atlas/sql/sqlclient"
-)
-
-// StatusReporter is used to gather information about migration status.
-type StatusReporter struct {
-	// Client configures the connection to the database to file a MigrateStatus for.
-	Client *sqlclient.Client
-	// DirURL of the migration directory.
-	DirURL *url.URL
-	// Dir is used for scanning and validating the migration directory.
-	Dir migrate.Dir
-	// Schema name the revision table resides in.
-	Schema string
-}
-
-// Report creates and writes a MigrateStatus.
-func (r *StatusReporter) Report(ctx context.Context) (*cmdlog.MigrateStatus, error) {
-	rep := &cmdlog.MigrateStatus{Env: cmdlog.NewEnv(r.Client, r.DirURL)}
-	// Check if there already is a revision table in the defined schema.
-	// Inspect schema and check if the table does already exist.
-	sch, err := r.Client.InspectSchema(ctx, r.Schema, &schema.InspectOptions{Tables: []string{revision.Table}})
-	if err != nil && !schema.IsNotExistError(err) {
-		return nil, err
-	}
-	if schema.IsNotExistError(err) || func() bool { _, ok := sch.Table(revision.Table); return !ok }() {
-		// Either schema or table does not exist.
-		if rep.Available, err = migrate.FilesFromLastCheckpoint(r.Dir); err != nil {
-			return nil, err
-		}
-		rep.Pending = rep.Available
-	} else {
-		// Both exist, fetch their data.
-		rrw, err := RevisionsForClient(ctx, r.Client, r.Schema)
-		if err != nil {
-			return nil, err
-		}
-		if err := rrw.Migrate(ctx); err != nil {
-			return nil, err
-		}
-		ex, err := migrate.NewExecutor(r.Client.Driver, r.Dir, rrw)
-		if err != nil {
-			return nil, err
-		}
-		rep.Applied, err = rrw.ReadRevisions(ctx)
-		if err != nil {
-			return nil, err
-		}
-		if rep.Pending, err = ex.Pending(ctx); err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
-			if err1 := (*migrate.HistoryNonLinearError)(nil); errors.As(err, &err1) {
-				rep.Error = err1.Error()
-				rep.Status = statusPending
-				rep.Pending = err1.Pending
-				rep.OutOfOrder = err1.OutOfOrder
-				// Non-linear error means at least one file was applied.
-				rep.Current = rep.Applied[len(rep.Applied)-1].Version
-				return rep, nil
-			}
-			return nil, err
-		}
-		// If no files were applied, all pending files are
-		// available. The first one might be a checkpoint.
-		if len(rep.Applied) == 0 {
-			rep.Available = rep.Pending
-		} else if rep.Available, err = r.Dir.Files(); err != nil {
-			return nil, err
-		}
-	}
-	switch len(rep.Pending) {
-	case len(rep.Available):
-		rep.Current = "No migration applied yet"
-	default:
-		rep.Current = rep.Applied[len(rep.Applied)-1].Version
-	}
-	if len(rep.Pending) == 0 {
-		rep.Status = statusOK
-		rep.Next = "Already at latest version"
-	} else {
-		rep.Status = statusPending
-		rep.Next = rep.Pending[0].Version()
-	}
-	// If the last one is partially applied (and not manually resolved).
-	if len(rep.Applied) != 0 {
-		last := rep.Applied[len(rep.Applied)-1]
-		if !last.Type.Has(migrate.RevisionTypeResolved) && last.Applied < last.Total {
-			rep.SQL = strings.ReplaceAll(last.ErrorStmt, "\n", " ")
-			rep.Error = strings.ReplaceAll(last.Error, "\n", " ")
-			rep.Count = last.Applied
-			idx := migrate.FilesLastIndex(rep.Available, func(f migrate.File) bool {
-				return f.Version() == last.Version
-			})
-			if idx == -1 {
-				return nil, fmt.Errorf("migration file with version %q not found", last.Version)
-			}
-			stmts, err := migrate.FileStmts(r.Client.Driver, rep.Available[idx])
-			if err != nil {
-				return nil, err
-			}
-			rep.Total = len(stmts)
-		}
-	}
-	return rep, nil
-}
-
-const (
-	statusOK      = "OK"
-	statusPending = "PENDING"
-)
diff --git a/sql/migrate/dir.go b/sql/migrate/dir.go
index d22265f3..6f0f89dc 100644
--- a/sql/migrate/dir.go
+++ b/sql/migrate/dir.go
@@ -561,6 +561,20 @@ func (t TemplateFormatter) Format(plan *Plan) ([]File, error) {
 	return files, nil
 }
 
+// FormatTo calls Format and writes the files' content to the given writer.
+func (t TemplateFormatter) FormatTo(plan *Plan, w io.Writer) error {
+	files, err := t.Format(plan)
+	if err != nil {
+		return err
+	}
+	for _, f := range files {
+		if _, err := w.Write(f.Bytes()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // HashFileName of the migration directory integrity sum file.
 const HashFileName = "atlas.sum"
 


Test patch:
diff --git a/cmd/atlas/internal/cmdlog/cmdlog_test.go b/cmd/atlas/internal/cmdlog/cmdlog_test.go
index 36a7ebd7..ccc6bb6f 100644
--- a/cmd/atlas/internal/cmdlog/cmdlog_test.go
+++ b/cmd/atlas/internal/cmdlog/cmdlog_test.go
@@ -5,9 +5,12 @@
 package cmdlog_test
 
 import (
+	cmdmigrate "ariga.io/atlas/cmd/atlas/internal/migrate"
 	"bytes"
 	"context"
 	"encoding/json"
+	"path/filepath"
+	"strings"
 	"testing"
 	"text/template"
 	"time"
@@ -631,3 +634,278 @@ func TestMigrateApply(t *testing.T) {
   -- 10ms
 `, b.String())
 }
+
+func TestReporter_Status(t *testing.T) {
+	var (
+		buf strings.Builder
+		ctx = context.Background()
+	)
+
+	// Clean.
+	dir, err := migrate.NewLocalDir(filepath.Join("../migrate/testdata", "broken"))
+	require.NoError(t, err)
+	c, err := sqlclient.Open(ctx, "sqlite://?mode=memory")
+	require.NoError(t, err)
+	defer c.Close()
+	rr := &cmdlog.StatusReporter{Client: c, Dir: dir}
+	report, err := rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: No migration applied yet
+  -- Next Version:    1
+  -- Executed Files:  0
+  -- Pending Files:   3
+`, buf.String())
+
+	// Applied one.
+	buf.Reset()
+	rrw, err := cmdmigrate.NewEntRevisions(ctx, c)
+	require.NoError(t, err)
+	require.NoError(t, rrw.Migrate(ctx))
+	ex, err := migrate.NewExecutor(c.Driver, dir, rrw)
+	require.NoError(t, err)
+	require.NoError(t, ex.ExecuteN(ctx, 1))
+	rr = &cmdlog.StatusReporter{Client: c, Dir: dir}
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 1
+  -- Next Version:    2
+  -- Executed Files:  1
+  -- Pending Files:   2
+`, buf.String())
+
+	// Applied two.
+	buf.Reset()
+	require.NoError(t, err)
+	require.NoError(t, ex.ExecuteN(ctx, 1))
+	rr = &cmdlog.StatusReporter{Client: c, Dir: dir}
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 2
+  -- Next Version:    3
+  -- Executed Files:  2
+  -- Pending Files:   1
+`, buf.String())
+
+	// Partial three.
+	buf.Reset()
+	require.NoError(t, err)
+	require.Error(t, ex.ExecuteN(ctx, 1))
+	rr = &cmdlog.StatusReporter{Client: c, Dir: dir}
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 3 (1 statements applied)
+  -- Next Version:    3 (1 statements left)
+  -- Executed Files:  3 (last one partially)
+  -- Pending Files:   1
+
+Last migration attempt had errors:
+  -- SQL:   THIS LINE ADDS A SYNTAX ERROR;
+  -- ERROR: near "THIS": syntax error
+`, buf.String())
+
+	// Fixed three - okay.
+	buf.Reset()
+	dir2, err := migrate.NewLocalDir(filepath.Join("../migrate/testdata", "fixed"))
+	require.NoError(t, err)
+	*dir = *dir2
+	require.NoError(t, err)
+	require.NoError(t, ex.ExecuteN(ctx, 1))
+	rr = &cmdlog.StatusReporter{Client: c, Dir: dir}
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: OK
+  -- Current Version: 3
+  -- Next Version:    Already at latest version
+  -- Executed Files:  3
+  -- Pending Files:   0
+`, buf.String())
+}
+
+func TestReporter_FromCheckpoint(t *testing.T) {
+	var (
+		buf strings.Builder
+		ctx = context.Background()
+	)
+	dir, err := migrate.NewLocalDir(t.TempDir())
+	require.NoError(t, err)
+	require.NoError(t, dir.WriteFile("1.sql", []byte("create table t1(c int);")))
+	require.NoError(t, dir.WriteFile("2.sql", []byte("create table t2(c int);")))
+	require.NoError(t, dir.WriteCheckpoint("3_checkpoint.sql", "", []byte("create table t1(c int);\ncreate table t2(c int);")))
+	sum, err := dir.Checksum()
+	require.NoError(t, err)
+	require.NoError(t, migrate.WriteSumFile(dir, sum))
+	require.NoError(t, migrate.Validate(dir))
+	c, err := sqlclient.Open(ctx, "sqlite://?mode=memory")
+	require.NoError(t, err)
+	defer c.Close()
+	rr := &cmdlog.StatusReporter{Client: c, Dir: dir}
+
+	// Clean.
+	report, err := rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: No migration applied yet
+  -- Next Version:    3 (checkpoint)
+  -- Executed Files:  0
+  -- Pending Files:   1
+`, buf.String())
+
+	// Clean and revisions table exists.
+	buf.Reset()
+	rrw, err := cmdmigrate.NewEntRevisions(ctx, c)
+	require.NoError(t, err)
+	require.NoError(t, rrw.Migrate(ctx))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: No migration applied yet
+  -- Next Version:    3 (checkpoint)
+  -- Executed Files:  0
+  -- Pending Files:   1
+`, buf.String())
+
+	// Execute one.
+	buf.Reset()
+	ex, err := migrate.NewExecutor(c.Driver, dir, rrw)
+	require.NoError(t, err)
+	require.NoError(t, ex.ExecuteN(ctx, 1))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: OK
+  -- Current Version: 3
+  -- Next Version:    Already at latest version
+  -- Executed Files:  1
+  -- Pending Files:   0
+`, buf.String())
+
+	// Add a new file.
+	buf.Reset()
+	require.NoError(t, dir.WriteFile("4.sql", []byte("create table t3(c int);")))
+	sum, err = dir.Checksum()
+	require.NoError(t, err)
+	require.NoError(t, migrate.WriteSumFile(dir, sum))
+	require.NoError(t, migrate.Validate(dir))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 3
+  -- Next Version:    4
+  -- Executed Files:  1
+  -- Pending Files:   1
+`, buf.String())
+
+	// Execute one.
+	buf.Reset()
+	require.NoError(t, ex.ExecuteN(ctx, 1))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: OK
+  -- Current Version: 4
+  -- Next Version:    Already at latest version
+  -- Executed Files:  2
+  -- Pending Files:   0
+`, buf.String())
+}
+
+func TestReporter_OutOfOrder(t *testing.T) {
+	var (
+		buf strings.Builder
+		ctx = context.Background()
+	)
+	dir, err := migrate.NewLocalDir(t.TempDir())
+	require.NoError(t, err)
+	require.NoError(t, dir.WriteFile("1.sql", []byte("create table t1(c int);")))
+	require.NoError(t, dir.WriteFile("2.sql", []byte("create table t2(c int);")))
+	sum, err := dir.Checksum()
+	require.NoError(t, err)
+	require.NoError(t, migrate.WriteSumFile(dir, sum))
+	c, err := sqlclient.Open(ctx, "sqlite://?mode=memory")
+	require.NoError(t, err)
+	defer c.Close()
+	rr := &cmdlog.StatusReporter{Client: c, Dir: dir}
+
+	rrw, err := cmdmigrate.NewEntRevisions(ctx, c)
+	require.NoError(t, err)
+	require.NoError(t, rrw.Migrate(ctx))
+	report, err := rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: No migration applied yet
+  -- Next Version:    1
+  -- Executed Files:  0
+  -- Pending Files:   2
+`, buf.String())
+
+	ex, err := migrate.NewExecutor(c.Driver, dir, rrw)
+	require.NoError(t, err)
+	require.NoError(t, ex.ExecuteN(ctx, 2))
+
+	// One file was added out of order.
+	buf.Reset()
+	require.NoError(t, dir.WriteFile("1.5.sql", []byte("create table t1_5(c int);")))
+	sum, err = dir.Checksum()
+	require.NoError(t, err)
+	require.NoError(t, migrate.WriteSumFile(dir, sum))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 2
+  -- Next Version:    UNKNOWN
+  -- Executed Files:  2
+  -- Pending Files:   1 (out of order)
+
+  ERROR: migration file 1.5.sql was added out of order. See: https://atlasgo.io/versioned/apply#non-linear-error
+`, buf.String())
+
+	// Multiple files were added our of order.
+	buf.Reset()
+	require.NoError(t, dir.WriteFile("1.6.sql", []byte("create table t1_6(c int);")))
+	sum, err = dir.Checksum()
+	require.NoError(t, err)
+	require.NoError(t, migrate.WriteSumFile(dir, sum))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 2
+  -- Next Version:    UNKNOWN
+  -- Executed Files:  2
+  -- Pending Files:   2 (out of order)
+
+  ERROR: migration files 1.5.sql, 1.6.sql were added out of order. See: https://atlasgo.io/versioned/apply#non-linear-error
+`, buf.String())
+
+	// A mix of pending and out of order files.
+	buf.Reset()
+	require.NoError(t, dir.WriteFile("3.sql", []byte("create table t3(c int);")))
+	sum, err = dir.Checksum()
+	require.NoError(t, err)
+	require.NoError(t, migrate.WriteSumFile(dir, sum))
+	report, err = rr.Report(ctx)
+	require.NoError(t, err)
+	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
+	require.Equal(t, `Migration Status: PENDING
+  -- Current Version: 2
+  -- Next Version:    UNKNOWN
+  -- Executed Files:  2
+  -- Pending Files:   3 (2 out of order)
+
+  ERROR: migration files 1.5.sql, 1.6.sql were added out of order. See: https://atlasgo.io/versioned/apply#non-linear-error
+`, buf.String())
+}
diff --git a/cmd/atlas/internal/migrate/report_test.go b/cmd/atlas/internal/migrate/report_test.go
deleted file mode 100644
index 612af2ef..00000000
--- a/cmd/atlas/internal/migrate/report_test.go
+++ /dev/null
@@ -1,294 +0,0 @@
-// Copyright 2021-present The Atlas Authors. All rights reserved.
-// This source code is licensed under the Apache 2.0 license found
-// in the LICENSE file in the root directory of this source tree.
-
-package migrate
-
-import (
-	"context"
-	"path/filepath"
-	"strings"
-	"testing"
-
-	"ariga.io/atlas/cmd/atlas/internal/cmdlog"
-	"ariga.io/atlas/sql/migrate"
-	"ariga.io/atlas/sql/sqlclient"
-
-	_ "github.com/mattn/go-sqlite3"
-	"github.com/stretchr/testify/require"
-)
-
-func TestReporter_Status(t *testing.T) {
-	var (
-		buf strings.Builder
-		ctx = context.Background()
-	)
-
-	// Clean.
-	dir, err := migrate.NewLocalDir(filepath.Join("../migrate/testdata", "broken"))
-	require.NoError(t, err)
-	c, err := sqlclient.Open(ctx, "sqlite://?mode=memory")
-	require.NoError(t, err)
-	defer c.Close()
-	rr := &StatusReporter{Client: c, Dir: dir}
-	report, err := rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: No migration applied yet
-  -- Next Version:    1
-  -- Executed Files:  0
-  -- Pending Files:   3
-`, buf.String())
-
-	// Applied one.
-	buf.Reset()
-	rrw, err := NewEntRevisions(ctx, c)
-	require.NoError(t, err)
-	require.NoError(t, rrw.Migrate(ctx))
-	ex, err := migrate.NewExecutor(c.Driver, dir, rrw)
-	require.NoError(t, err)
-	require.NoError(t, ex.ExecuteN(ctx, 1))
-	rr = &StatusReporter{Client: c, Dir: dir}
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 1
-  -- Next Version:    2
-  -- Executed Files:  1
-  -- Pending Files:   2
-`, buf.String())
-
-	// Applied two.
-	buf.Reset()
-	require.NoError(t, err)
-	require.NoError(t, ex.ExecuteN(ctx, 1))
-	rr = &StatusReporter{Client: c, Dir: dir}
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 2
-  -- Next Version:    3
-  -- Executed Files:  2
-  -- Pending Files:   1
-`, buf.String())
-
-	// Partial three.
-	buf.Reset()
-	require.NoError(t, err)
-	require.Error(t, ex.ExecuteN(ctx, 1))
-	rr = &StatusReporter{Client: c, Dir: dir}
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 3 (1 statements applied)
-  -- Next Version:    3 (1 statements left)
-  -- Executed Files:  3 (last one partially)
-  -- Pending Files:   1
-
-Last migration attempt had errors:
-  -- SQL:   THIS LINE ADDS A SYNTAX ERROR;
-  -- ERROR: near "THIS": syntax error
-`, buf.String())
-
-	// Fixed three - okay.
-	buf.Reset()
-	dir2, err := migrate.NewLocalDir(filepath.Join("../migrate/testdata", "fixed"))
-	require.NoError(t, err)
-	*dir = *dir2
-	require.NoError(t, err)
-	require.NoError(t, ex.ExecuteN(ctx, 1))
-	rr = &StatusReporter{Client: c, Dir: dir}
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: OK
-  -- Current Version: 3
-  -- Next Version:    Already at latest version
-  -- Executed Files:  3
-  -- Pending Files:   0
-`, buf.String())
-}
-
-func TestReporter_FromCheckpoint(t *testing.T) {
-	var (
-		buf strings.Builder
-		ctx = context.Background()
-	)
-	dir, err := migrate.NewLocalDir(t.TempDir())
-	require.NoError(t, err)
-	require.NoError(t, dir.WriteFile("1.sql", []byte("create table t1(c int);")))
-	require.NoError(t, dir.WriteFile("2.sql", []byte("create table t2(c int);")))
-	require.NoError(t, dir.WriteCheckpoint("3_checkpoint.sql", "", []byte("create table t1(c int);\ncreate table t2(c int);")))
-	sum, err := dir.Checksum()
-	require.NoError(t, err)
-	require.NoError(t, migrate.WriteSumFile(dir, sum))
-	require.NoError(t, migrate.Validate(dir))
-	c, err := sqlclient.Open(ctx, "sqlite://?mode=memory")
-	require.NoError(t, err)
-	defer c.Close()
-	rr := &StatusReporter{Client: c, Dir: dir}
-
-	// Clean.
-	report, err := rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: No migration applied yet
-  -- Next Version:    3 (checkpoint)
-  -- Executed Files:  0
-  -- Pending Files:   1
-`, buf.String())
-
-	// Clean and revisions table exists.
-	buf.Reset()
-	rrw, err := NewEntRevisions(ctx, c)
-	require.NoError(t, err)
-	require.NoError(t, rrw.Migrate(ctx))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: No migration applied yet
-  -- Next Version:    3 (checkpoint)
-  -- Executed Files:  0
-  -- Pending Files:   1
-`, buf.String())
-
-	// Execute one.
-	buf.Reset()
-	ex, err := migrate.NewExecutor(c.Driver, dir, rrw)
-	require.NoError(t, err)
-	require.NoError(t, ex.ExecuteN(ctx, 1))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: OK
-  -- Current Version: 3
-  -- Next Version:    Already at latest version
-  -- Executed Files:  1
-  -- Pending Files:   0
-`, buf.String())
-
-	// Add a new file.
-	buf.Reset()
-	require.NoError(t, dir.WriteFile("4.sql", []byte("create table t3(c int);")))
-	sum, err = dir.Checksum()
-	require.NoError(t, err)
-	require.NoError(t, migrate.WriteSumFile(dir, sum))
-	require.NoError(t, migrate.Validate(dir))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 3
-  -- Next Version:    4
-  -- Executed Files:  1
-  -- Pending Files:   1
-`, buf.String())
-
-	// Execute one.
-	buf.Reset()
-	require.NoError(t, ex.ExecuteN(ctx, 1))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: OK
-  -- Current Version: 4
-  -- Next Version:    Already at latest version
-  -- Executed Files:  2
-  -- Pending Files:   0
-`, buf.String())
-}
-
-func TestReporter_OutOfOrder(t *testing.T) {
-	var (
-		buf strings.Builder
-		ctx = context.Background()
-	)
-	dir, err := migrate.NewLocalDir(t.TempDir())
-	require.NoError(t, err)
-	require.NoError(t, dir.WriteFile("1.sql", []byte("create table t1(c int);")))
-	require.NoError(t, dir.WriteFile("2.sql", []byte("create table t2(c int);")))
-	sum, err := dir.Checksum()
-	require.NoError(t, err)
-	require.NoError(t, migrate.WriteSumFile(dir, sum))
-	c, err := sqlclient.Open(ctx, "sqlite://?mode=memory")
-	require.NoError(t, err)
-	defer c.Close()
-	rr := &StatusReporter{Client: c, Dir: dir}
-
-	rrw, err := NewEntRevisions(ctx, c)
-	require.NoError(t, err)
-	require.NoError(t, rrw.Migrate(ctx))
-	report, err := rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: No migration applied yet
-  -- Next Version:    1
-  -- Executed Files:  0
-  -- Pending Files:   2
-`, buf.String())
-
-	ex, err := migrate.NewExecutor(c.Driver, dir, rrw)
-	require.NoError(t, err)
-	require.NoError(t, ex.ExecuteN(ctx, 2))
-
-	// One file was added out of order.
-	buf.Reset()
-	require.NoError(t, dir.WriteFile("1.5.sql", []byte("create table t1_5(c int);")))
-	sum, err = dir.Checksum()
-	require.NoError(t, err)
-	require.NoError(t, migrate.WriteSumFile(dir, sum))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 2
-  -- Next Version:    UNKNOWN
-  -- Executed Files:  2
-  -- Pending Files:   1 (out of order)
-
-  ERROR: migration file 1.5.sql was added out of order. See: https://atlasgo.io/versioned/apply#non-linear-error
-`, buf.String())
-
-	// Multiple files were added our of order.
-	buf.Reset()
-	require.NoError(t, dir.WriteFile("1.6.sql", []byte("create table t1_6(c int);")))
-	sum, err = dir.Checksum()
-	require.NoError(t, err)
-	require.NoError(t, migrate.WriteSumFile(dir, sum))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 2
-  -- Next Version:    UNKNOWN
-  -- Executed Files:  2
-  -- Pending Files:   2 (out of order)
-
-  ERROR: migration files 1.5.sql, 1.6.sql were added out of order. See: https://atlasgo.io/versioned/apply#non-linear-error
-`, buf.String())
-
-	// A mix of pending and out of order files.
-	buf.Reset()
-	require.NoError(t, dir.WriteFile("3.sql", []byte("create table t3(c int);")))
-	sum, err = dir.Checksum()
-	require.NoError(t, err)
-	require.NoError(t, migrate.WriteSumFile(dir, sum))
-	report, err = rr.Report(ctx)
-	require.NoError(t, err)
-	require.NoError(t, cmdlog.MigrateStatusTemplate.Execute(&buf, report))
-	require.Equal(t, `Migration Status: PENDING
-  -- Current Version: 2
-  -- Next Version:    UNKNOWN
-  -- Executed Files:  2
-  -- Pending Files:   3 (2 out of order)
-
-  ERROR: migration files 1.5.sql, 1.6.sql were added out of order. See: https://atlasgo.io/versioned/apply#non-linear-error
-`, buf.String())
-}
diff --git a/sql/migrate/dir_test.go b/sql/migrate/dir_test.go
index f9e47bbf..f2d06cfe 100644
--- a/sql/migrate/dir_test.go
+++ b/sql/migrate/dir_test.go
@@ -531,6 +531,21 @@ func TestDirTar(t *testing.T) {
 	require.Equal(t, "create table t(c int);", string(files[0].Bytes()))
 }
 
+func TestDefaultFormatter_FormatTo(t *testing.T) {
+	var b bytes.Buffer
+	err := migrate.DefaultFormatter.FormatTo(&migrate.Plan{
+		Changes: []*migrate.Change{
+			{Cmd: "create table t1(c int)"},
+			{Cmd: "create table t2(c int)", Comment: "create table"},
+		},
+	}, &b)
+	require.NoError(t, err)
+	require.Equal(t, `create table t1(c int);
+-- Create table
+create table t2(c int);
+`, b.String())
+}
+
 func fileNames(r io.Reader) ([]string, error) {
 	var out []string
 	tr := tar.NewReader(r)