- conga: TestOutcome*, TestCancelQueuedNeverRuns, TestCancelRunningCancelsCtx, TestStopJobFromWorker, TestManagerPHPSemantics, principal, delay, registration, worker and queue-setting branches; TestQueueClear and TestQueueWork move to commands_test.go with the batch/state and queue-filter cases (coverage 92.5%) - scheduler: validation, ordering, missing catalog, log writer, dueAt - lagoon: TestQueueMigrationsUpDown (River v7 + summer_jobs, rollback, idempotent rerun), OnDatabase isolation, Transaction edges - bonfire TestCallEdges, pact TestCadence
147 lines
4.7 KiB
Go
147 lines
4.7 KiB
Go
package lagoon
|
|
|
|
import (
|
|
"context"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/riverqueue/river/rivermigrate"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
func riverTables(t *testing.T, gdb *gorm.DB) []string {
|
|
t.Helper()
|
|
var names []string
|
|
if err := gdb.Raw(`SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename LIKE 'river\_%' ORDER BY tablename`).Scan(&names).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return names
|
|
}
|
|
|
|
func tableExists(t *testing.T, gdb *gorm.DB, name string) bool {
|
|
t.Helper()
|
|
var n int
|
|
if err := gdb.Raw(`SELECT count(*) FROM pg_tables WHERE schemaname = 'public' AND tablename = ?`, name).Scan(&n).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return n == 1
|
|
}
|
|
|
|
// TestQueueMigrationsUpDown covers the summercms.conga migration set on a
|
|
// fresh database: River's schema pinned at version 7 and summer_jobs with
|
|
// the WinterCMS apparatus columns plus river_job_id; rolling the set back
|
|
// one step at a time removes summer_jobs and then River's schema; running
|
|
// Migrate again restores both and a further run is a no-op.
|
|
func TestQueueMigrationsUpDown(t *testing.T) {
|
|
db, _ := dedicatedDB(t, "lagoon_queue_migrations")
|
|
gdb, err := Use(t.Context(), db)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := Migrate(gdb, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// River v7's table set, read from pg_tables after migrating (11-01).
|
|
wantRiver := []string{"river_job", "river_leader", "river_migration", "river_notification", "river_queue"}
|
|
if got := riverTables(t, gdb); !reflect.DeepEqual(got, wantRiver) {
|
|
t.Fatalf("river tables = %v, want %v", got, wantRiver)
|
|
}
|
|
var version int
|
|
if err := gdb.Raw(`SELECT max(version) FROM river_migration`).Scan(&version).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if version != RiverSchemaVersion {
|
|
t.Fatalf("river schema version = %d, want %d", version, RiverSchemaVersion)
|
|
}
|
|
|
|
type column struct {
|
|
Name string `gorm:"column:column_name"`
|
|
Type string `gorm:"column:data_type"`
|
|
Nullable string `gorm:"column:is_nullable"`
|
|
Default *string
|
|
}
|
|
var cols []column
|
|
if err := gdb.Raw(`SELECT column_name, data_type, is_nullable, column_default AS "default"
|
|
FROM information_schema.columns WHERE table_name = ? ORDER BY ordinal_position`, JobsTable).Scan(&cols).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := []struct {
|
|
name, typ, nullable, def string
|
|
}{
|
|
{"id", "integer", "NO", "nextval"},
|
|
{"label", "character varying", "NO", ""},
|
|
{"status", "integer", "NO", "0"},
|
|
{"progress", "integer", "NO", "0"},
|
|
{"progress_max", "integer", "NO", "0"},
|
|
{"user_id", "integer", "YES", ""},
|
|
{"is_admin", "boolean", "NO", "false"},
|
|
{"is_canceled", "boolean", "NO", "false"},
|
|
{"metadata", "text", "NO", ""},
|
|
{"river_job_id", "bigint", "YES", ""},
|
|
{"created_at", "timestamp with time zone", "YES", ""},
|
|
{"updated_at", "timestamp with time zone", "YES", ""},
|
|
}
|
|
if len(cols) != len(want) {
|
|
t.Fatalf("summer_jobs has %d columns, want %d: %+v", len(cols), len(want), cols)
|
|
}
|
|
for i, w := range want {
|
|
c := cols[i]
|
|
def := ""
|
|
if c.Default != nil {
|
|
def = *c.Default
|
|
}
|
|
if c.Name != w.name || c.Type != w.typ || c.Nullable != w.nullable || !strings.HasPrefix(def, w.def) || (w.def == "" && def != "") {
|
|
t.Errorf("column %d = %s %s nullable=%s default=%q, want %s %s nullable=%s default %q", i, c.Name, c.Type, c.Nullable, def, w.name, w.typ, w.nullable, w.def)
|
|
}
|
|
}
|
|
|
|
sqlDB, err := gdb.DB()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
m, err := migrator(gdb, QueueHistoryID, QueueMigrations(sqlDB))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := m.RollbackLast(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if tableExists(t, gdb, JobsTable) {
|
|
t.Fatal("summer_jobs survived the rollback of its migration")
|
|
}
|
|
if !tableExists(t, gdb, "river_job") {
|
|
t.Fatal("rolling back summer_jobs removed River's schema")
|
|
}
|
|
if err := m.RollbackLast(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if left := riverTables(t, gdb); len(left) != 0 {
|
|
t.Fatalf("River tables left after the rollback: %v", left)
|
|
}
|
|
|
|
if err := Migrate(gdb, nil); err != nil {
|
|
t.Fatalf("migrate after rollback: %v", err)
|
|
}
|
|
if !tableExists(t, gdb, JobsTable) || !tableExists(t, gdb, "river_job") {
|
|
t.Fatal("migrate after rollback did not restore the queue tables")
|
|
}
|
|
if err := Migrate(gdb, nil); err != nil {
|
|
t.Fatalf("second migrate is not idempotent: %v", err)
|
|
}
|
|
var history int
|
|
if err := gdb.Raw(`SELECT count(*) FROM summer_migrations_summercms_conga`).Scan(&history).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if history != 2 {
|
|
t.Fatalf("history rows = %d, want 2", history)
|
|
}
|
|
|
|
if err := migrateRiver(context.Background(), nil, rivermigrate.DirectionUp, nil); err == nil {
|
|
t.Fatal("migrateRiver accepted a nil pool")
|
|
}
|
|
if txContext(nil) == nil || txContext(&gorm.DB{}) == nil {
|
|
t.Fatal("txContext returned a nil context")
|
|
}
|
|
}
|