relspecgo/pkg/writers/pgsql/live_execute_test.go

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