package pgsql import ( "context" "encoding/json" "fmt" "os" "path/filepath" "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_WriteDatabaseFailedStatementContinues(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}) if err != nil { t.Fatalf("failed statements must not abort the run: %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 _. 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 { 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) } }