refactor(backend): migrate test infrastructure to isolated databases
Add CreateTestDatabase function for parallel isolated test databases per package. - Add CreateTestDatabase() for isolated test DBs (parallel-safe) - Move all TestMain functions to per-package testmain_test.go files - Remove old TestMain from handlers_test.go, jwt_test.go, main_test.go - Add JWT init guard in main.go to skip when -test.* flags detected - Update testdb.go with admin DSN and proper cleanup - Rename test database to crussell_test_db for consistency - Replace testdb.NewPool + testdb.Migrate pattern with CreateTestDatabase Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
Sisyphus
+157
-163
@@ -6,17 +6,131 @@ package testdb
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
const defaultTestDSN = "postgres://myuser:mypassword@localhost:5432/crussell_test?sslmode=disable&require_auth=scram-sha-256"
|
||||
const (
|
||||
defaultTestDSN = "postgres://myuser:mypassword@localhost:5432/crussell_test?sslmode=disable"
|
||||
adminDatabaseDSN = "postgres://myuser:mypassword@localhost:5432/mydb?sslmode=disable"
|
||||
)
|
||||
|
||||
// CreateTestDatabase creates a fresh isolated test database, runs the schema migration,
|
||||
// and returns a pool connected to it. If a database with the same name exists, it's
|
||||
// dropped first. This enables parallel test execution across packages since each
|
||||
// package gets its own database running migrations in parallel.
|
||||
func CreateTestDatabase(dbName string) *pgxpool.Pool {
|
||||
ctx := context.Background()
|
||||
|
||||
adminPool, err := pgxpool.New(ctx, adminDatabaseDSN)
|
||||
if err != nil {
|
||||
log.Fatalf("testdb: failed to connect to admin database: %v", err)
|
||||
}
|
||||
defer adminPool.Close()
|
||||
|
||||
safeName := pgx.Identifier{dbName}.Sanitize()
|
||||
|
||||
// Drop existing database with retries (terminate connections first)
|
||||
for attempt := 0; attempt < 3; attempt++ {
|
||||
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`
|
||||
SELECT pg_terminate_backend(pid)
|
||||
FROM pg_stat_activity
|
||||
WHERE datname = '%s' AND pid != pg_backend_pid()
|
||||
`, dbName))
|
||||
|
||||
_, err = adminPool.Exec(ctx, fmt.Sprintf("DROP DATABASE IF EXISTS %s", safeName))
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
log.Printf("testdb: retry %d dropping %s: %v", attempt+1, dbName, err)
|
||||
}
|
||||
if err != nil {
|
||||
log.Fatalf("testdb: failed to drop database %s after retries: %v", dbName, err)
|
||||
}
|
||||
|
||||
_, err = adminPool.Exec(ctx, fmt.Sprintf("CREATE DATABASE %s", safeName))
|
||||
if err != nil {
|
||||
log.Fatalf("testdb: failed to create database %s: %v", dbName, err)
|
||||
}
|
||||
adminPool.Close()
|
||||
|
||||
testDSN := fmt.Sprintf("postgres://myuser:mypassword@localhost:5432/%s?sslmode=disable", dbName)
|
||||
pool, err := pgxpool.New(ctx, testDSN)
|
||||
if err != nil {
|
||||
log.Fatalf("testdb: failed to connect to %s: %v", dbName, err)
|
||||
}
|
||||
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
log.Fatalf("testdb: failed to ping %s: %v", dbName, err)
|
||||
}
|
||||
|
||||
if err := migratePool(pool); err != nil {
|
||||
log.Fatalf("testdb: migration failed: %v", err)
|
||||
}
|
||||
|
||||
return pool
|
||||
}
|
||||
|
||||
// DestroyTestDatabase closes the pool (if non-nil) and drops the test database.
|
||||
// Retries aggressively with exponential backoff, pg_terminate_backend between
|
||||
// attempts, and a final DROP DATABASE WITH (FORCE) on PostgreSQL 13+.
|
||||
func DestroyTestDatabase(pool *pgxpool.Pool, dbName string) {
|
||||
if dbName == "" {
|
||||
return
|
||||
}
|
||||
|
||||
if pool != nil {
|
||||
pool.Close()
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
adminPool, err := pgxpool.New(ctx, adminDatabaseDSN)
|
||||
if err != nil {
|
||||
log.Printf("testdb: warning: failed to connect to drop %s: %v", dbName, err)
|
||||
return
|
||||
}
|
||||
defer adminPool.Close()
|
||||
|
||||
safeName := pgx.Identifier{dbName}.Sanitize()
|
||||
|
||||
// Aggressive retry with pg_terminate_backend between attempts and final FORCE option.
|
||||
for attempt := 0; attempt < 10; attempt++ {
|
||||
// Kill all connections to the target database.
|
||||
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`
|
||||
SELECT pg_terminate_backend(pid)
|
||||
FROM pg_stat_activity
|
||||
WHERE datname = '%s' AND pid != pg_backend_pid()
|
||||
`, dbName))
|
||||
|
||||
// Try DROP DATABASE WITH (FORCE) first (PG13+) — this kills remaining connections atomically.
|
||||
_, err = adminPool.Exec(ctx, fmt.Sprintf("DROP DATABASE IF EXISTS %s WITH (FORCE)", safeName))
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Fallback to plain DROP (may work if pg_terminate_backend was sufficient).
|
||||
_, err = adminPool.Exec(ctx, fmt.Sprintf("DROP DATABASE IF EXISTS %s", safeName))
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Exponential backoff: 100ms, 200ms, 400ms, 800ms, 1.6s, 3.2s, 6.4s, 12.8s, 25.6s, 51.2s
|
||||
wait := time.Duration(100*(1<<attempt)) * time.Millisecond
|
||||
log.Printf("testdb: retry %d dropping %s: %v (retrying in %v)", attempt+1, dbName, err, wait)
|
||||
time.Sleep(wait)
|
||||
}
|
||||
|
||||
log.Printf("testdb: warning: failed to drop database %s after 10 attempts: %v", dbName, err)
|
||||
}
|
||||
|
||||
// Pool creates a new test database pool using TEST_DB_DSN env var or default DSN.
|
||||
func Pool(t *testing.T) *pgxpool.Pool {
|
||||
t.Helper()
|
||||
|
||||
@@ -56,116 +170,23 @@ func NewPool(dsn string) (*pgxpool.Pool, error) {
|
||||
return pool, nil
|
||||
}
|
||||
|
||||
func Migrate(t *testing.T, pool *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
// migratePool executes the full schema migration on the given pool.
|
||||
// This internal version returns errors directly for use by CreateTestDatabase.
|
||||
func migratePool(pool *pgxpool.Pool) error {
|
||||
ctx := context.Background()
|
||||
|
||||
// Acquire a dedicated connection so advisory lock, migration, and unlock all use the same session
|
||||
conn, err := pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to acquire connection for migration: %v", err)
|
||||
return fmt.Errorf("failed to acquire connection for migration: %w", err)
|
||||
}
|
||||
defer conn.Release()
|
||||
return migrateSchema(ctx, conn)
|
||||
}
|
||||
|
||||
_, err = conn.Exec(ctx, "SELECT pg_advisory_lock(1337)")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to acquire migration advisory lock: %v", err)
|
||||
}
|
||||
defer conn.Exec(ctx, "SELECT pg_advisory_unlock(1337)")
|
||||
|
||||
// Check if database already has tables or types
|
||||
var typeCount int
|
||||
err = conn.QueryRow(ctx, "SELECT COUNT(*) FROM pg_type WHERE typnamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'public')").Scan(&typeCount)
|
||||
if err == nil && typeCount > 0 {
|
||||
// Drop TYPES FIRST (they have CASCADE dependencies on tables)
|
||||
typeDrops := []string{
|
||||
"DROP TYPE IF EXISTS account_role CASCADE",
|
||||
"DROP TYPE IF EXISTS account_type CASCADE",
|
||||
"DROP TYPE IF EXISTS payment_type CASCADE",
|
||||
"DROP TYPE IF EXISTS payment_method CASCADE",
|
||||
"DROP TYPE IF EXISTS payment_status CASCADE",
|
||||
"DROP TYPE IF EXISTS booking_status CASCADE",
|
||||
"DROP TYPE IF EXISTS verification_purpose CASCADE",
|
||||
"DROP TYPE IF EXISTS admin_notification_reason CASCADE",
|
||||
"DROP TYPE IF EXISTS campaign_type CASCADE",
|
||||
"DROP TYPE IF EXISTS milestone_type CASCADE",
|
||||
"DROP TYPE IF EXISTS milestone_unit CASCADE",
|
||||
"DROP TYPE IF EXISTS discount_campaign_scope CASCADE",
|
||||
"DROP TYPE IF EXISTS discount_campaign_status CASCADE",
|
||||
"DROP TYPE IF EXISTS till_item_type CASCADE",
|
||||
}
|
||||
for _, stmt := range typeDrops {
|
||||
if _, err := conn.Exec(ctx, stmt); err != nil {
|
||||
t.Logf("Warning dropping type: %v (expected if using IF EXISTS)", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Drop all tables, sequences, and views in correct order
|
||||
dropOrder := []string{
|
||||
"till_sales",
|
||||
"admin_audit_log",
|
||||
"gift_card_transactions",
|
||||
"gift_card_expired_balances",
|
||||
"booking_discounts",
|
||||
"loyalty_redemptions",
|
||||
"discount_campaigns",
|
||||
"forgiven_no_shows",
|
||||
"admin_notifications",
|
||||
"user_notification_preferences",
|
||||
"user_referrals",
|
||||
"booking_custom_services",
|
||||
"booking_services",
|
||||
"refunds",
|
||||
"payments",
|
||||
"user_saved_cards",
|
||||
"square_deposits",
|
||||
"affiliate_payouts",
|
||||
"financial_aggregates",
|
||||
"gift_cards",
|
||||
"user_giftcard_balances",
|
||||
"bookings",
|
||||
"user_patch_tests",
|
||||
"patch_tests",
|
||||
"booking_edit_requests",
|
||||
"services",
|
||||
"custom_services",
|
||||
"verification_codes",
|
||||
"user_social_logins",
|
||||
"users",
|
||||
"images",
|
||||
"tags",
|
||||
"time_blockers",
|
||||
"working_hours",
|
||||
"exceptional_group_applications",
|
||||
"exceptional_working_hours",
|
||||
"exceptional_working_hours_groups",
|
||||
"business_settings",
|
||||
"login_audit",
|
||||
"refresh_tokens",
|
||||
"revoked_jtis",
|
||||
}
|
||||
|
||||
for _, table := range dropOrder {
|
||||
stmt := fmt.Sprintf("DROP TABLE IF EXISTS %s CASCADE", table)
|
||||
if _, err := conn.Exec(ctx, stmt); err != nil {
|
||||
t.Logf("Warning dropping table %s: %v (expected if using IF EXISTS)", table, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Drop sequences last
|
||||
seqDrops := []string{
|
||||
"DROP SEQUENCE IF EXISTS invoice_number_seq",
|
||||
"DROP SEQUENCE IF EXISTS tags_id_seq",
|
||||
"DROP SEQUENCE IF EXISTS exceptional_working_hours_id_seq",
|
||||
"DROP SEQUENCE IF EXISTS exceptional_working_hours_groups_id_seq",
|
||||
}
|
||||
for _, stmt := range seqDrops {
|
||||
if _, err := conn.Exec(ctx, stmt); err != nil {
|
||||
t.Logf("Warning dropping sequence: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// migrateSchema executes the schema migration on an existing connection.
|
||||
// Returns an error if any step fails.
|
||||
// NOTE: This is always called on a fresh database (created via CreateTestDatabase),
|
||||
// so no defensive cleanup of types/tables/sequences is needed.
|
||||
func migrateSchema(ctx context.Context, conn *pgxpool.Conn) error {
|
||||
paths := []string{
|
||||
"../../../init-scripts/init-script.sql",
|
||||
"../../init-scripts/init-script.sql",
|
||||
@@ -182,7 +203,7 @@ func Migrate(t *testing.T, pool *pgxpool.Pool) {
|
||||
}
|
||||
|
||||
if schemaSQL == "" {
|
||||
t.Fatal("Could not find init-script.sql in any expected location")
|
||||
return fmt.Errorf("could not find init-script.sql in any expected location")
|
||||
}
|
||||
|
||||
// Split and execute the schema SQL statement-by-statement for robust execution and diagnostic visibility
|
||||
@@ -197,11 +218,19 @@ func Migrate(t *testing.T, pool *pgxpool.Pool) {
|
||||
firstLine = firstLine[:80] + "..."
|
||||
}
|
||||
|
||||
_, err = conn.Exec(ctx, stmt)
|
||||
_, err := conn.Exec(ctx, stmt)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to execute migration statement [%d]: %s\nError: %v", i+1, firstLine, err)
|
||||
return fmt.Errorf("failed to execute migration statement [%d]: %s\nError: %w", i+1, firstLine, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func Migrate(t *testing.T, pool *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
if err := migratePool(pool); err != nil {
|
||||
t.Fatal(err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func splitSQLStatements(sql string) []string {
|
||||
@@ -263,68 +292,33 @@ func TruncateTables(t *testing.T, pool *pgxpool.Pool) {
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Acquire a dedicated connection so advisory lock, truncations, and unlock all use the same session
|
||||
conn, err := pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to acquire connection for truncation: %v", err)
|
||||
}
|
||||
defer conn.Release()
|
||||
|
||||
_, err = conn.Exec(ctx, "SELECT pg_advisory_lock(1338)")
|
||||
// Single TRUNCATE with all tables — one round-trip instead of ~44.
|
||||
// CASCADE handles FK dependencies so order doesn't matter.
|
||||
// Advisory lock removed: each package has its own database now.
|
||||
_, err = conn.Exec(ctx, `
|
||||
TRUNCATE TABLE
|
||||
till_sales, admin_audit_log, gift_card_transactions,
|
||||
gift_card_expired_balances, booking_discounts, loyalty_redemptions,
|
||||
discount_campaigns, forgiven_no_shows, name_history, referral_discounts,
|
||||
user_social_logins, verification_codes, booking_custom_services,
|
||||
booking_services, refunds, payments, user_saved_cards, square_deposits,
|
||||
affiliate_payouts, financial_aggregates, gift_cards, user_giftcard_balances,
|
||||
bookings, booking_edit_requests, user_patch_tests, patch_tests, services,
|
||||
custom_services, admin_notifications, user_referrals,
|
||||
user_notification_preferences, time_blockers, working_hours,
|
||||
exceptional_group_applications, exceptional_working_hours,
|
||||
exceptional_working_hours_groups, business_settings, users, images, tags,
|
||||
login_audit, refresh_tokens, revoked_jtis
|
||||
CASCADE
|
||||
`)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to acquire truncate advisory lock: %v", err)
|
||||
}
|
||||
defer conn.Exec(ctx, "SELECT pg_advisory_unlock(1338)")
|
||||
|
||||
tables := []string{
|
||||
"till_sales",
|
||||
"admin_audit_log",
|
||||
"gift_card_transactions",
|
||||
"gift_card_expired_balances",
|
||||
"booking_discounts",
|
||||
"loyalty_redemptions",
|
||||
"discount_campaigns",
|
||||
"forgiven_no_shows",
|
||||
"user_social_logins",
|
||||
"verification_codes",
|
||||
"booking_custom_services",
|
||||
"booking_services",
|
||||
"refunds",
|
||||
"payments",
|
||||
"user_saved_cards",
|
||||
"square_deposits",
|
||||
"affiliate_payouts",
|
||||
"financial_aggregates",
|
||||
"gift_cards",
|
||||
"user_giftcard_balances",
|
||||
"bookings",
|
||||
"booking_edit_requests",
|
||||
"user_patch_tests",
|
||||
"patch_tests",
|
||||
"services",
|
||||
"custom_services",
|
||||
"admin_notifications",
|
||||
"user_referrals",
|
||||
"user_notification_preferences",
|
||||
"time_blockers",
|
||||
"working_hours",
|
||||
"exceptional_group_applications",
|
||||
"exceptional_working_hours",
|
||||
"exceptional_working_hours_groups",
|
||||
"business_settings",
|
||||
"users",
|
||||
"images",
|
||||
"tags",
|
||||
"login_audit",
|
||||
"refresh_tokens",
|
||||
"revoked_jtis",
|
||||
}
|
||||
|
||||
for _, table := range tables {
|
||||
_, err := conn.Exec(ctx, fmt.Sprintf("TRUNCATE TABLE %s CASCADE", table))
|
||||
if err != nil {
|
||||
t.Logf("Warning: could not truncate %s: %v", table, err)
|
||||
}
|
||||
t.Logf("Warning: truncation failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user