418 lines
15 KiB
Go
418 lines
15 KiB
Go
package pgsql
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"git.warky.dev/wdevs/relspecgo/pkg/models"
|
|
"git.warky.dev/wdevs/relspecgo/pkg/writers"
|
|
)
|
|
|
|
func liveWriterConn(t *testing.T) string {
|
|
t.Helper()
|
|
conn := os.Getenv("RELSPEC_TEST_PG_CONN")
|
|
if conn == "" {
|
|
t.Skip("RELSPEC_TEST_PG_CONN not set")
|
|
}
|
|
return conn
|
|
}
|
|
|
|
// liveWriterSchema returns a unique schema name and drops it on cleanup.
|
|
func liveWriterSchema(t *testing.T, connString string) (string, *pgx.Conn) {
|
|
t.Helper()
|
|
ctx := context.Background()
|
|
conn, err := pgx.Connect(ctx, connString)
|
|
if err != nil {
|
|
t.Fatalf("connect: %v", err)
|
|
}
|
|
name := fmt.Sprintf("pgw_test_%d", time.Now().UnixNano())
|
|
t.Cleanup(func() {
|
|
_, _ = conn.Exec(ctx, "DROP SCHEMA IF EXISTS "+name+" CASCADE")
|
|
_ = conn.Close(ctx)
|
|
})
|
|
return name, conn
|
|
}
|
|
|
|
func liveModel(schemaName string, columns map[string]string) *models.Database {
|
|
db := models.InitDatabase("live")
|
|
s := models.InitSchema(schemaName)
|
|
tbl := models.InitTable("accounts", schemaName)
|
|
id := models.InitColumn("id", "accounts", schemaName)
|
|
id.Type = "integer"
|
|
id.NotNull = true
|
|
id.IsPrimaryKey = true
|
|
tbl.Columns["id"] = id
|
|
for name, typ := range columns {
|
|
c := models.InitColumn(name, "accounts", schemaName)
|
|
c.Type = typ
|
|
tbl.Columns[name] = c
|
|
}
|
|
s.Tables = append(s.Tables, tbl)
|
|
db.Schemas = append(db.Schemas, s)
|
|
return db
|
|
}
|
|
|
|
func runLiveWrite(t *testing.T, connString string, db *models.Database, meta map[string]interface{}) (*ExecutionReport, error) {
|
|
t.Helper()
|
|
m := map[string]interface{}{"connection_string": connString}
|
|
for k, v := range meta {
|
|
m[k] = v
|
|
}
|
|
w := NewWriter(&writers.WriterOptions{Metadata: m})
|
|
err := w.WriteDatabase(db)
|
|
return w.executionReport, err
|
|
}
|
|
|
|
func columnExists(t *testing.T, conn *pgx.Conn, schema, table, column string) bool {
|
|
t.Helper()
|
|
var ok bool
|
|
err := conn.QueryRow(context.Background(),
|
|
`SELECT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=$1 AND table_name=$2 AND column_name=$3)`,
|
|
schema, table, column).Scan(&ok)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return ok
|
|
}
|
|
|
|
func TestLive_WriteDatabaseEmptyThenIdenticalThenDrifted(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
reportPath := filepath.Join(t.TempDir(), "report.json")
|
|
meta := map[string]interface{}{"report_path": reportPath}
|
|
|
|
// Empty database: schema and table are created.
|
|
rep, err := runLiveWrite(t, connString, liveModel(schema, map[string]string{"name": "text"}), meta)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if rep.FailedStatements != 0 || rep.ExecutedStatements == 0 {
|
|
t.Fatalf("first run report: %+v", rep)
|
|
}
|
|
if !columnExists(t, conn, schema, "accounts", "name") {
|
|
t.Fatal("column name not created")
|
|
}
|
|
data, err := os.ReadFile(reportPath)
|
|
if err != nil {
|
|
t.Fatalf("report not written: %v", err)
|
|
}
|
|
var onDisk ExecutionReport
|
|
if err := json.Unmarshal(data, &onDisk); err != nil || onDisk.TotalStatements != rep.TotalStatements {
|
|
t.Errorf("report on disk mismatch: %v %+v", err, onDisk)
|
|
}
|
|
created := false
|
|
for _, s := range rep.Schemas {
|
|
for _, tb := range s.Tables {
|
|
if tb.Name == "accounts" && tb.Created {
|
|
created = true
|
|
}
|
|
}
|
|
}
|
|
if !created {
|
|
t.Errorf("table creation not tracked: %+v", rep.Schemas)
|
|
}
|
|
|
|
// Identical database: nothing to execute.
|
|
rep, err = runLiveWrite(t, connString, liveModel(schema, map[string]string{"name": "text"}), nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if rep.TotalStatements != 0 {
|
|
t.Errorf("identical DB must produce no statements, got %d", rep.TotalStatements)
|
|
}
|
|
|
|
// Drifted database: only the new column is added.
|
|
rep, err = runLiveWrite(t, connString, liveModel(schema, map[string]string{"name": "text", "email": "text"}), nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if rep.FailedStatements != 0 || rep.TotalStatements == 0 {
|
|
t.Errorf("drift report: %+v", rep)
|
|
}
|
|
if !columnExists(t, conn, schema, "accounts", "email") {
|
|
t.Error("drifted column email not added")
|
|
}
|
|
}
|
|
|
|
func TestLive_WriteDatabaseFailedStatementReturnsError(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, _ := liveWriterSchema(t, connString)
|
|
reportPath := filepath.Join(t.TempDir(), "report.json")
|
|
|
|
db := liveModel(schema, map[string]string{"bad": "no_such_type_xyz"})
|
|
rep, err := runLiveWrite(t, connString, db, map[string]interface{}{"full_ddl": true, "report_path": reportPath})
|
|
var failure *ExecutionFailure
|
|
if !errors.As(err, &failure) {
|
|
t.Fatalf("expected structured failure, got: %v", err)
|
|
}
|
|
if rep.FailedStatements == 0 || len(rep.Errors) != rep.FailedStatements {
|
|
t.Fatalf("expected recorded failures: %+v", rep)
|
|
}
|
|
e := rep.Errors[0]
|
|
if e.StatementNumber == 0 || e.Statement == "" || e.Error == "" {
|
|
t.Errorf("incomplete error entry: %+v", e)
|
|
}
|
|
if _, err := os.Stat(reportPath); err != nil {
|
|
t.Errorf("report must be written even on failures: %v", err)
|
|
}
|
|
failedTable := false
|
|
for _, s := range rep.Schemas {
|
|
for _, tb := range s.Tables {
|
|
if tb.Name == "accounts" && !tb.Created && tb.Error != "" {
|
|
failedTable = true
|
|
}
|
|
}
|
|
}
|
|
if !failedTable {
|
|
t.Errorf("failed table creation not tracked: %+v", rep.Schemas)
|
|
}
|
|
}
|
|
|
|
func TestLive_WriteDatabaseFlattenFallsBackToFullDDL(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
|
|
w := NewWriter(&writers.WriterOptions{
|
|
FlattenSchema: true,
|
|
Metadata: map[string]interface{}{"connection_string": connString},
|
|
})
|
|
if err := w.WriteDatabase(liveModel(schema, nil)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Flattened output lands in public as <schema>_<table>.
|
|
flat := "public." + schema + "_accounts"
|
|
t.Cleanup(func() { _, _ = conn.Exec(context.Background(), "DROP TABLE IF EXISTS "+flat+" CASCADE") })
|
|
var ok bool
|
|
if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", flat).Scan(&ok); err != nil || !ok {
|
|
t.Errorf("flattened table %s not created (ok=%v err=%v)", flat, ok, err)
|
|
}
|
|
}
|
|
|
|
func TestGenerateLiveDiffStatements_FlattenRejected(t *testing.T) {
|
|
w := NewWriter(&writers.WriterOptions{FlattenSchema: true})
|
|
if _, err := w.generateLiveDiffStatements(models.InitDatabase("x"), "postgres://unused"); err == nil {
|
|
t.Error("flatten must be rejected before connecting")
|
|
}
|
|
}
|
|
|
|
func TestExecuteStatements_ConnectFailure(t *testing.T) {
|
|
w := NewWriter(&writers.WriterOptions{})
|
|
w.executionReport = &ExecutionReport{}
|
|
err := w.executeStatements([]string{"SELECT 1"}, "postgres://nobody:x@127.0.0.1:1/none?connect_timeout=1")
|
|
if err == nil {
|
|
t.Error("expected connect failure")
|
|
}
|
|
if w.executionReport.TotalStatements != 1 {
|
|
t.Errorf("total not recorded: %+v", w.executionReport)
|
|
}
|
|
}
|
|
|
|
func TestLive_ExecuteStatementsSkipsCommentsAndBlank(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
|
|
w := NewWriter(&writers.WriterOptions{})
|
|
w.executionReport = &ExecutionReport{}
|
|
stmts := []string{
|
|
"-- Schema: " + schema,
|
|
" ",
|
|
"CREATE SCHEMA " + schema,
|
|
"CREATE TABLE " + schema + ".t (id int)",
|
|
"-- plain comment",
|
|
}
|
|
if err := w.executeStatements(stmts, connString); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r := w.executionReport
|
|
if r.ExecutedStatements != 2 || r.FailedStatements != 0 || r.TotalStatements != 5 || r.SkippedStatements != 3 {
|
|
t.Errorf("counts: %+v", r)
|
|
}
|
|
if len(r.Schemas) != 1 || r.Schemas[0].Name != schema || len(r.Schemas[0].Tables) != 1 || !r.Schemas[0].Tables[0].Created {
|
|
t.Errorf("schema tracking: %+v", r.Schemas)
|
|
}
|
|
var ok bool
|
|
if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", schema+".t").Scan(&ok); err != nil || !ok {
|
|
t.Errorf("table not created: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestLive_ExecuteStatementsFailureModes(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
for _, continueOnError := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("continue=%v", continueOnError), func(t *testing.T) {
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
reportPath := filepath.Join(t.TempDir(), "report.json")
|
|
w := NewWriter(&writers.WriterOptions{
|
|
ContinueOnError: continueOnError,
|
|
Metadata: map[string]interface{}{"report_path": reportPath},
|
|
})
|
|
w.executionReport = &ExecutionReport{}
|
|
err := w.executeStatements([]string{
|
|
"CREATE SCHEMA " + schema,
|
|
"SELECT 1 / 0",
|
|
"CREATE TABLE " + schema + ".after_failure (id int)",
|
|
}, connString)
|
|
var failure *ExecutionFailure
|
|
if !errors.As(err, &failure) || failure.Report != w.executionReport {
|
|
t.Fatalf("expected structured execution failure, got %v", err)
|
|
}
|
|
var report ExecutionReport
|
|
data, err := os.ReadFile(reportPath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := json.Unmarshal(data, &report); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
executed, skipped := 1, 1
|
|
if continueOnError {
|
|
executed, skipped = 2, 0
|
|
}
|
|
if report.TotalStatements != 3 || report.ExecutedStatements != executed || report.FailedStatements != 1 || report.SkippedStatements != skipped || report.EndTime == "" {
|
|
t.Fatalf("incorrect partial application report: %+v", report)
|
|
}
|
|
if len(report.Errors) != 1 || report.Errors[0].StatementNumber != 2 {
|
|
t.Fatalf("incorrect failure details: %+v", report.Errors)
|
|
}
|
|
if exists := columnExists(t, conn, schema, "after_failure", "id"); exists != continueOnError {
|
|
t.Fatalf("later statement executed=%v, continue=%v", exists, continueOnError)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestLive_IncompatibleTypeChangePreservesColumnAndData(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
ctx := context.Background()
|
|
_, err := conn.Exec(ctx, "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text UNIQUE); INSERT INTO "+schema+".accounts VALUES (1, 'not-an-integer')")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var originalIndex uint32
|
|
if err := conn.QueryRow(ctx, "SELECT to_regclass($1)::oid", schema+".accounts_name_key").Scan(&originalIndex); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Exercise both full-DDL and diff generation against populated data.
|
|
for _, fullDDL := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("full_ddl=%v", fullDDL), func(t *testing.T) {
|
|
model := liveModel(schema, map[string]string{"name": "integer"})
|
|
unique := models.InitConstraint("accounts_name_key", models.UniqueConstraint)
|
|
unique.Columns = []string{"name"}
|
|
model.Schemas[0].Tables[0].Constraints[unique.Name] = unique
|
|
report, err := runLiveWrite(t, connString, model, map[string]interface{}{"full_ddl": fullDDL})
|
|
var failure *ExecutionFailure
|
|
if !errors.As(err, &failure) || report.FailedStatements != 1 {
|
|
t.Fatalf("expected failed conversion, got report=%+v err=%v", report, err)
|
|
}
|
|
var value, typ string
|
|
if err := conn.QueryRow(ctx, "SELECT name, pg_typeof(name)::text FROM "+schema+".accounts WHERE id=1").Scan(&value, &typ); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if value != "not-an-integer" || typ != "text" {
|
|
t.Fatalf("original column changed: value=%q type=%q", value, typ)
|
|
}
|
|
var columns int
|
|
if err := conn.QueryRow(ctx, "SELECT count(*) FROM information_schema.columns WHERE table_schema=$1 AND table_name='accounts'", schema).Scan(&columns); err != nil || columns != 2 {
|
|
t.Fatalf("unexpected replacement column: columns=%d err=%v", columns, err)
|
|
}
|
|
var index uint32
|
|
if err := conn.QueryRow(ctx, "SELECT to_regclass($1)::oid", schema+".accounts_name_key").Scan(&index); err != nil || index != originalIndex {
|
|
t.Fatalf("unique index changed: index=%d original=%d err=%v", index, originalIndex, err)
|
|
}
|
|
if _, err := conn.Exec(ctx, "INSERT INTO "+schema+".accounts VALUES (2, 'not-an-integer')"); err == nil {
|
|
t.Fatal("unique constraint no longer enforced")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestLive_AtomicExecutionRollback(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
w := NewWriter(&writers.WriterOptions{AtomicExecution: true})
|
|
w.executionReport = &ExecutionReport{}
|
|
err := w.executeStatements([]string{
|
|
"-- Schema: " + schema,
|
|
"CREATE SCHEMA " + schema,
|
|
"CREATE TABLE " + schema + ".created (id int)",
|
|
"SELECT 1 / 0",
|
|
"SELECT 1",
|
|
}, connString)
|
|
var failure *ExecutionFailure
|
|
if !errors.As(err, &failure) {
|
|
t.Fatalf("expected execution failure: %v", err)
|
|
}
|
|
report := failure.Report
|
|
if report.Outcome != "rolled_back" || report.RolledBackStatements != 2 || report.ExecutedStatements != 0 || report.FailedStatements != 1 || report.SkippedStatements != 2 {
|
|
t.Fatalf("rollback report: %+v", report)
|
|
}
|
|
if report.Schemas[0].Tables[0].Created {
|
|
t.Fatal("rolled-back table must not be reported created")
|
|
}
|
|
if columnExists(t, conn, schema, "created", "id") {
|
|
t.Fatal("table survived transaction rollback")
|
|
}
|
|
}
|
|
|
|
func TestLive_TypePreflightPrecedesDropsEvenWithContinuation(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
_, err := conn.Exec(context.Background(), "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text UNIQUE); INSERT INTO "+schema+".accounts VALUES (1, '123'), (2, 'bad')")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// The desired model would drop the unique constraint. Incompatible data must
|
|
// be caught first, including bad data after the first row.
|
|
model := liveModel(schema, map[string]string{"name": "integer"})
|
|
w := NewWriter(&writers.WriterOptions{ContinueOnError: true, Metadata: map[string]interface{}{"connection_string": connString}})
|
|
err = w.WriteDatabase(model)
|
|
var failure *ExecutionFailure
|
|
if !errors.As(err, &failure) || failure.Report.FailedStatements != 1 || failure.Report.ExecutedStatements != 0 {
|
|
t.Fatalf("preflight must stop mutations: %+v %v", w.executionReport, err)
|
|
}
|
|
var exists bool
|
|
if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", schema+".accounts_name_key").Scan(&exists); err != nil || !exists {
|
|
t.Fatalf("constraint was dropped before preflight: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestLive_CustomTypeConversionAndPredicate(t *testing.T) {
|
|
connString := liveWriterConn(t)
|
|
for _, fullDDL := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("full_ddl=%v", fullDDL), func(t *testing.T) {
|
|
schema, conn := liveWriterSchema(t, connString)
|
|
_, err := conn.Exec(context.Background(), "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text); INSERT INTO "+schema+".accounts VALUES (1,'42 units'),(2,'7 units')")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
model := liveModel(schema, map[string]string{"name": "integer"})
|
|
col := model.Schemas[0].Tables[0].Columns["name"]
|
|
col.Metadata["pgsql_using"] = "split_part(name, ' ', 1)::integer"
|
|
col.Metadata["pgsql_preflight"] = "name ~ '^[0-9]+ units$' AND name NOT LIKE '7 %'"
|
|
meta := map[string]interface{}{"full_ddl": fullDDL}
|
|
if _, err := runLiveWrite(t, connString, model, meta); err == nil || !strings.Contains(err.Error(), "preflight rejected rows") {
|
|
t.Fatalf("predicate must reject: %v", err)
|
|
}
|
|
col.Metadata["pgsql_preflight"] = "name ~ '^[0-9]+ units$'"
|
|
if _, err := runLiveWrite(t, connString, model, meta); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var value int
|
|
if err := conn.QueryRow(context.Background(), "SELECT sum(name) FROM "+schema+".accounts").Scan(&value); err != nil || value != 49 {
|
|
t.Fatalf("custom conversion result %d: %v", value, err)
|
|
}
|
|
})
|
|
}
|
|
}
|