relspecgo/pkg/writers/sqlexec/writer_live_test.go

package sqlexec

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"

	"git.warky.dev/wdevs/relspecgo/pkg/assetloader"
	"git.warky.dev/wdevs/relspecgo/pkg/models"
	"git.warky.dev/wdevs/relspecgo/pkg/writers"
)

func TestWriter_Options(t *testing.T) {
	opts := &writers.WriterOptions{Metadata: map[string]interface{}{"k": "v"}}
	if got := NewWriter(opts).Options(); got != opts {
		t.Error("Options must return the same pointer")
	}
}

func TestWriter_ConnectFailure(t *testing.T) {
	opts := &writers.WriterOptions{Metadata: map[string]interface{}{
		"connection_string": "postgres://nobody:nopass@127.0.0.1:1/none?connect_timeout=1",
	}}
	w := NewWriter(opts)
	scripts := []*models.Script{{Name: "s", SQL: "SELECT 1"}}

	if err := w.WriteDatabase(&models.Database{Schemas: []*models.Schema{{Name: "public", Scripts: scripts}}}); err == nil ||
		!strings.Contains(err.Error(), "failed to connect") {
		t.Errorf("WriteDatabase: %v", err)
	}
	if err := w.WriteSchema(&models.Schema{Name: "public", Scripts: scripts}); err == nil ||
		!strings.Contains(err.Error(), "failed to connect") {
		t.Errorf("WriteSchema: %v", err)
	}
}

// liveConn returns a connection string for a live PostgreSQL or skips the test.
func liveConn(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
}

// liveSchema creates a throwaway schema and drops it on cleanup.
func liveSchema(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("sqlexec_test_%d", time.Now().UnixNano())
	if _, err := conn.Exec(ctx, "CREATE SCHEMA "+name); err != nil {
		t.Fatalf("create schema: %v", err)
	}
	t.Cleanup(func() {
		_, _ = conn.Exec(ctx, "DROP SCHEMA IF EXISTS "+name+" CASCADE")
		_ = conn.Close(ctx)
	})
	return name, conn
}

func liveOptions(connString string, extra map[string]interface{}) *writers.WriterOptions {
	meta := map[string]interface{}{"connection_string": connString}
	for k, v := range extra {
		meta[k] = v
	}
	return &writers.WriterOptions{Metadata: meta}
}

func TestLive_ExecuteScriptsOrder(t *testing.T) {
	connString := liveConn(t)
	schema, conn := liveSchema(t, connString)
	ctx := context.Background()

	// Each script appends its own name; the resulting row order is the execution order.
	mk := func(name string, prio int, seq uint) *models.Script {
		return &models.Script{
			Name: name, Priority: prio, Sequence: seq,
			SQL: fmt.Sprintf("INSERT INTO %s.log(name) VALUES ('%s');", schema, name),
		}
	}
	scripts := []*models.Script{
		{Name: "00_create", Priority: 0, SQL: fmt.Sprintf("CREATE TABLE %s.log(id serial primary key, name text);", schema)},
		mk("c_late", 2, 1),
		mk("b_prio1_seq2", 1, 2),
		mk("a_prio1_seq1", 1, 1),
		mk("a_same", 1, 3),
		mk("b_same", 1, 3),
		{Name: "empty", Priority: 1, Sequence: 0, SQL: ""},
	}

	opts := liveOptions(connString, nil)
	if err := NewWriter(opts).WriteSchema(&models.Schema{Name: schema, Scripts: scripts}); err != nil {
		t.Fatalf("WriteSchema: %v", err)
	}

	rows, err := conn.Query(ctx, fmt.Sprintf("SELECT name FROM %s.log ORDER BY id", schema))
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	var got []string
	for rows.Next() {
		var n string
		if err := rows.Scan(&n); err != nil {
			t.Fatal(err)
		}
		got = append(got, n)
	}
	want := []string{"a_prio1_seq1", "b_prio1_seq2", "a_same", "b_same", "c_late"}
	if strings.Join(got, ",") != strings.Join(want, ",") {
		t.Errorf("execution order = %v, want %v", got, want)
	}
	if opts.Metadata["execution_total"] != 6 || opts.Metadata["execution_success"] != 6 || opts.Metadata["execution_failed"] != 0 {
		t.Errorf("counts: %v", opts.Metadata)
	}
}

func TestLive_FailingScriptStops(t *testing.T) {
	connString := liveConn(t)
	schema, conn := liveSchema(t, connString)
	ctx := context.Background()

	scripts := []*models.Script{
		{Name: "01_ok", Priority: 1, SQL: fmt.Sprintf("CREATE TABLE %s.a(id int);", schema)},
		{Name: "02_bad", Priority: 2, SQL: "SELECT * FROM definitely_missing_table;"},
		{Name: "03_never", Priority: 3, SQL: fmt.Sprintf("CREATE TABLE %s.never(id int);", schema)},
	}
	err := NewWriter(liveOptions(connString, nil)).WriteSchema(&models.Schema{Name: schema, Scripts: scripts})
	if err == nil || !strings.Contains(err.Error(), "02_bad") {
		t.Fatalf("expected failure naming 02_bad, got %v", err)
	}

	var exists bool
	if err := conn.QueryRow(ctx, "SELECT to_regclass($1) IS NOT NULL", schema+".never").Scan(&exists); err != nil {
		t.Fatal(err)
	}
	if exists {
		t.Error("script after the failure must not run")
	}
}

func TestLive_IgnoreErrorsContinues(t *testing.T) {
	connString := liveConn(t)
	schema, conn := liveSchema(t, connString)
	ctx := context.Background()

	scripts := []*models.Script{
		{Name: "01_bad", Priority: 1, SQL: "SELECT * FROM definitely_missing_table;"},
		{Name: "02_ok", Priority: 2, SQL: fmt.Sprintf("CREATE TABLE %s.after(id int);", schema)},
	}
	opts := liveOptions(connString, map[string]interface{}{"ignore_errors": true})
	if err := NewWriter(opts).WriteSchema(&models.Schema{Name: schema, Scripts: scripts}); err != nil {
		t.Fatalf("ignore_errors must not fail: %v", err)
	}
	if opts.Metadata["execution_total"] != 2 || opts.Metadata["execution_success"] != 1 || opts.Metadata["execution_failed"] != 1 {
		t.Errorf("counts: %v", opts.Metadata)
	}
	var exists bool
	if err := conn.QueryRow(ctx, "SELECT to_regclass($1) IS NOT NULL", schema+".after").Scan(&exists); err != nil || !exists {
		t.Errorf("later script must run: exists=%v err=%v", exists, err)
	}
}

func TestLive_EmbedDirectiveErrorHandling(t *testing.T) {
	connString := liveConn(t)
	schema, _ := liveSchema(t, connString)

	bad := models.InitScript("embed_bad")
	bad.Priority = 1
	bad.SQL = "-- @embed: path=missing.txt var=:body mode=text\nSELECT :body;"
	bad.Metadata[assetloader.ScriptSourcePathMetadataKey] = filepath.Join(t.TempDir(), "s.sql")
	if err := NewWriter(liveOptions(connString, nil)).WriteSchema(&models.Schema{Name: schema, Scripts: []*models.Script{bad}}); err == nil ||
		!strings.Contains(err.Error(), "embed_bad") {
		t.Errorf("expected error naming script, got %v", err)
	}

	opts := liveOptions(connString, map[string]interface{}{"ignore_errors": true})
	if err := NewWriter(opts).WriteSchema(&models.Schema{Name: schema, Scripts: []*models.Script{bad}}); err != nil {
		t.Errorf("ignore_errors: %v", err)
	}
	if opts.Metadata["execution_failed"] != 1 {
		t.Errorf("counts: %v", opts.Metadata)
	}
}

func TestLive_WriteDatabaseMultiSchema(t *testing.T) {
	connString := liveConn(t)
	s1, conn := liveSchema(t, connString)
	s2, _ := liveSchema(t, connString)
	ctx := context.Background()

	db := &models.Database{Schemas: []*models.Schema{
		{Name: s1, Scripts: []*models.Script{{Name: "a", SQL: fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.t(id int);", s1)}}},
		{Name: s2, Scripts: []*models.Script{{Name: "b", SQL: fmt.Sprintf("CREATE TABLE %s.t(id int);", s2)}}},
	}}
	if err := NewWriter(liveOptions(connString, nil)).WriteDatabase(db); err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{s1, s2} {
		var ok bool
		if err := conn.QueryRow(ctx, "SELECT to_regclass($1) IS NOT NULL", s+".t").Scan(&ok); err != nil || !ok {
			t.Errorf("table in %s missing (err %v)", s, err)
		}
	}

	// A failure in one schema aborts and names that schema.
	db.Schemas[1].Scripts[0].SQL = "SELECT * FROM definitely_missing_table;"
	err := NewWriter(liveOptions(connString, nil)).WriteDatabase(db)
	if err == nil || !strings.Contains(err.Error(), "schema "+s2) {
		t.Errorf("expected error naming schema %s, got %v", s2, err)
	}
}