From 6bfef0fcf5e46c404a716ca8ac7133752b5a8865 Mon Sep 17 00:00:00 2001 From: youfak Date: Thu, 30 Jul 2026 10:35:55 +0800 Subject: [PATCH] test: add postgres admin state contract fixture --- .github/workflows/ci.yml | 16 ++ deploy/compose_test.go | 19 +++ .../contract_integration_test.go | 83 ++++++++++ internal/adapters/postgresadmin/migrate.go | 47 ++++++ .../adapters/postgresadmin/migrations_test.go | 17 +++ .../rollback_integration_test.go | 142 ++++++++++++++++++ .../postgresadmin/testpostgres_test.go | 98 ++++++++++++ scripts/test-postgres.ps1 | 34 +++++ 8 files changed, 456 insertions(+) create mode 100644 internal/adapters/postgresadmin/contract_integration_test.go create mode 100644 internal/adapters/postgresadmin/migrate.go create mode 100644 internal/adapters/postgresadmin/rollback_integration_test.go create mode 100644 internal/adapters/postgresadmin/testpostgres_test.go create mode 100644 scripts/test-postgres.ps1 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e0829a1..c6d4433 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,3 +32,19 @@ jobs: go-version-file: go.mod cache: true - run: go test -race -timeout 60s ./internal/... + + integration: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true + - name: Redis activity contract + shell: pwsh + run: ./scripts/test-redis.ps1 + - name: PostgreSQL admin-state contract + shell: pwsh + run: ./scripts/test-postgres.ps1 diff --git a/deploy/compose_test.go b/deploy/compose_test.go index f7f099e..e84cc34 100644 --- a/deploy/compose_test.go +++ b/deploy/compose_test.go @@ -57,6 +57,25 @@ func TestRedisFixtureScriptUsesDedicatedComposeProject(t *testing.T) { } } +func TestPostgresFixtureScriptUsesDedicatedComposeProject(t *testing.T) { + payload, err := os.ReadFile("../scripts/test-postgres.ps1") + if err != nil { + t.Fatalf("read test-postgres.ps1: %v", err) + } + script := string(payload) + for _, required := range []string{ + `-p $composeProject`, + `up -d --wait --wait-timeout 60 postgres`, + `PROXY_POOL_TEST_POSTGRES_URL`, + `go test -count=1 -tags=integration -timeout 60s ./internal/adapters/postgresadmin/...`, + `down --volumes --remove-orphans`, + } { + if !strings.Contains(script, required) { + t.Errorf("test-postgres.ps1 missing %q", required) + } + } +} + func TestLocalRedisIsExplicitlyEphemeral(t *testing.T) { document := loadComposeDocument(t) redis, ok := document.Services["redis"] diff --git a/internal/adapters/postgresadmin/contract_integration_test.go b/internal/adapters/postgresadmin/contract_integration_test.go new file mode 100644 index 0000000..3658b74 --- /dev/null +++ b/internal/adapters/postgresadmin/contract_integration_test.go @@ -0,0 +1,83 @@ +//go:build integration + +package postgresadmin + +import ( + "context" + "slices" + "strings" + "testing" + + "proxy-pool/internal/domain/adminstate" + "proxy-pool/internal/domain/adminstate/contracttest" +) + +func TestPostgresAdminStateContract(t *testing.T) { + contracttest.Run(t, func(t *testing.T) adminstate.Store { + return newPostgresTestFixture(t).Store + }) +} + +func TestMigrationsAreIdempotentAndBounded(t *testing.T) { + fixture := newPostgresTestFixture(t) + if err := ApplyMigrations(t.Context(), fixture.Pool); err != nil { + t.Fatalf("ApplyMigrations(second run): %v", err) + } + + rows, err := fixture.Pool.Query(context.Background(), ` + SELECT table_name + FROM information_schema.tables + WHERE table_schema = $1 AND table_type = 'BASE TABLE' + ORDER BY table_name`, fixture.Schema) + if err != nil { + t.Fatalf("query migrated tables: %v", err) + } + defer rows.Close() + tables := make([]string, 0, 6) + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + t.Fatalf("scan migrated table: %v", err) + } + tables = append(tables, name) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate migrated tables: %v", err) + } + want := []string{ + "admin_audit_log", + "admin_outbox", + "config_revisions", + "control_revisions", + "routing_admin_state", + "upstream_admin_state", + } + if !slices.Equal(tables, want) { + t.Fatalf("migrated tables = %v, want %v", tables, want) + } + + columnRows, err := fixture.Pool.Query(context.Background(), ` + SELECT table_name, column_name + FROM information_schema.columns + WHERE table_schema = $1`, fixture.Schema) + if err != nil { + t.Fatalf("query migrated columns: %v", err) + } + defer columnRows.Close() + for columnRows.Next() { + var table string + var column string + if err := columnRows.Scan(&table, &column); err != nil { + t.Fatalf("scan migrated column: %v", err) + } + qualified := strings.ToLower(table + "." + column) + for _, forbidden := range []string{"proxy", "credential", "extraction", "worker_owner", "idempotency"} { + if strings.Contains(qualified, forbidden) { + t.Fatalf("PostgreSQL management boundary contains forbidden column %s", qualified) + } + } + } + if err := columnRows.Err(); err != nil { + t.Fatalf("iterate migrated columns: %v", err) + } +} diff --git a/internal/adapters/postgresadmin/migrate.go b/internal/adapters/postgresadmin/migrate.go new file mode 100644 index 0000000..70410ce --- /dev/null +++ b/internal/adapters/postgresadmin/migrate.go @@ -0,0 +1,47 @@ +package postgresadmin + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "proxy-pool/internal/domain/adminstate" +) + +// ApplyMigrations runs every embedded idempotent migration on one physical connection. +func ApplyMigrations(ctx context.Context, pool *pgxpool.Pool) error { + if err := contextError(ctx); err != nil { + return err + } + if isNil(pool) { + return adminstate.ErrInvalidCommand + } + connection, err := pool.Acquire(ctx) + if err != nil { + return migrationError(ctx, err) + } + defer connection.Release() + + for _, migration := range Migrations() { + if _, err := connection.Exec(ctx, migration.SQL); err != nil { + rollbackContext, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _ = connection.Exec(rollbackContext, "ROLLBACK") + return migrationError(ctx, err) + } + } + return nil +} + +func migrationError(ctx context.Context, err error) error { + if ctx != nil && ctx.Err() != nil { + return ctx.Err() + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return err + } + return fmt.Errorf("postgresadmin: apply migrations: %w", adminstate.ErrUnavailable) +} diff --git a/internal/adapters/postgresadmin/migrations_test.go b/internal/adapters/postgresadmin/migrations_test.go index 44d96f0..b59fc63 100644 --- a/internal/adapters/postgresadmin/migrations_test.go +++ b/internal/adapters/postgresadmin/migrations_test.go @@ -1,12 +1,29 @@ package postgresadmin import ( + "context" + "errors" "regexp" "sort" "strings" "testing" + + "proxy-pool/internal/domain/adminstate" ) +func TestApplyMigrationsValidatesContextAndPool(t *testing.T) { + t.Parallel() + + if err := ApplyMigrations(context.Background(), nil); !errors.Is(err, adminstate.ErrInvalidCommand) { + t.Fatalf("ApplyMigrations(nil) error = %v, want ErrInvalidCommand", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := ApplyMigrations(ctx, nil); !errors.Is(err, context.Canceled) { + t.Fatalf("ApplyMigrations(canceled) error = %v, want context.Canceled", err) + } +} + func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) { t.Parallel() migrations := Migrations() diff --git a/internal/adapters/postgresadmin/rollback_integration_test.go b/internal/adapters/postgresadmin/rollback_integration_test.go new file mode 100644 index 0000000..8adb580 --- /dev/null +++ b/internal/adapters/postgresadmin/rollback_integration_test.go @@ -0,0 +1,142 @@ +//go:build integration + +package postgresadmin + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "proxy-pool/internal/domain/adminstate" +) + +func TestMutationRollsBackWhenAuditInsertFails(t *testing.T) { + fixture := newPostgresTestFixture(t) + now := integrationNow() + commitIntegrationConfig(t, fixture.Store, now) + installRejectInsertTrigger(t, fixture, "admin_audit_log", "reject_admin_audit") + + _, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-audit-failure", Actor: integrationActor(), OccurredAt: now.Add(time.Second), + Name: "provider-a", Enabled: false, + }) + if !errors.Is(err, adminstate.ErrUnavailable) { + t.Fatalf("SetUpstreamEnabled(audit failure) error = %v, want ErrUnavailable", err) + } + assertFailedMutationLeftBaseline(t, fixture) + + dropRejectInsertTrigger(t, fixture, "admin_audit_log", "reject_admin_audit") + result, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-after-audit-failure", Actor: integrationActor(), OccurredAt: now.Add(2 * time.Second), + Name: "provider-a", Enabled: false, + }) + if err != nil || !result.Changed || result.Revision != 2 { + t.Fatalf("SetUpstreamEnabled(after rollback) = %+v, %v", result, err) + } +} + +func TestMutationRollsBackWhenOutboxInsertFails(t *testing.T) { + fixture := newPostgresTestFixture(t) + now := integrationNow() + commitIntegrationConfig(t, fixture.Store, now) + installRejectInsertTrigger(t, fixture, "admin_outbox", "reject_admin_outbox") + + _, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-outbox-failure", Actor: integrationActor(), OccurredAt: now.Add(time.Second), + Name: "provider-a", Enabled: false, + }) + if !errors.Is(err, adminstate.ErrUnavailable) { + t.Fatalf("SetUpstreamEnabled(outbox failure) error = %v, want ErrUnavailable", err) + } + assertFailedMutationLeftBaseline(t, fixture) + + dropRejectInsertTrigger(t, fixture, "admin_outbox", "reject_admin_outbox") + result, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-after-outbox-failure", Actor: integrationActor(), OccurredAt: now.Add(2 * time.Second), + Name: "provider-a", Enabled: false, + }) + if err != nil || !result.Changed || result.Revision != 2 { + t.Fatalf("SetUpstreamEnabled(after rollback) = %+v, %v", result, err) + } +} + +func assertFailedMutationLeftBaseline(t *testing.T, fixture postgresTestFixture) { + t.Helper() + snapshot, err := fixture.Store.Snapshot(context.Background()) + if err != nil || snapshot.Revision != 1 || !integrationUpstreamEnabled(snapshot, "provider-a") { + t.Fatalf("Snapshot(after failed mutation) = %+v, %v", snapshot, err) + } + audits, err := fixture.Store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10}) + if err != nil || len(audits) != 1 { + t.Fatalf("ReadAudit(after failed mutation) = %+v, %v", audits, err) + } + for table, want := range map[string]int{"control_revisions": 1, "admin_outbox": 1} { + var count int + if err := fixture.Pool.QueryRow(context.Background(), "SELECT COUNT(*) FROM "+table).Scan(&count); err != nil { + t.Fatalf("count %s: %v", table, err) + } + if count != want { + t.Fatalf("%s row count = %d, want %d", table, count, want) + } + } +} + +func installRejectInsertTrigger(t *testing.T, fixture postgresTestFixture, table, trigger string) { + t.Helper() + function := trigger + "_fn" + statement := "CREATE FUNCTION " + function + `() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'injected management write failure'; END + $$; + CREATE TRIGGER ` + trigger + " BEFORE INSERT ON " + table + + " FOR EACH ROW EXECUTE FUNCTION " + function + "()" + if _, err := fixture.Pool.Exec(context.Background(), statement); err != nil { + t.Fatalf("install %s trigger: %v", trigger, err) + } +} + +func dropRejectInsertTrigger(t *testing.T, fixture postgresTestFixture, table, trigger string) { + t.Helper() + statement := "DROP TRIGGER " + trigger + " ON " + table + "; DROP FUNCTION " + trigger + "_fn()" + if _, err := fixture.Pool.Exec(context.Background(), statement); err != nil { + t.Fatalf("drop %s trigger: %v", trigger, err) + } +} + +func commitIntegrationConfig(t *testing.T, store adminstate.Store, now time.Time) { + t.Helper() + result, err := store.CommitConfig(context.Background(), adminstate.CommitConfigCommand{ + RequestID: "req-config", Actor: integrationActor(), OccurredAt: now, + ConfigVersion: "cfg-1", Checksum: strings.Repeat("a", adminstate.SHA256HexBytes), + Source: "configs/proxy-pool.yaml", + Upstreams: []adminstate.UpstreamDefinition{ + {Name: "provider-a", Enabled: true}, + {Name: "provider-b", Enabled: true}, + }, + Routings: []adminstate.RoutingDefinition{{ + Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"}, + CurrentUpstream: "provider-a", + }}, + }) + if err != nil || !result.Changed || result.Revision != 1 { + t.Fatalf("CommitConfig() = %+v, %v", result, err) + } +} + +func integrationActor() adminstate.Actor { + return adminstate.Actor{ID: "admin-a", SourceIP: "192.0.2.10"} +} + +func integrationNow() time.Time { + return time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC) +} + +func integrationUpstreamEnabled(snapshot adminstate.Snapshot, name string) bool { + for _, upstream := range snapshot.Upstreams { + if upstream.Name == name { + return upstream.Enabled + } + } + return false +} diff --git a/internal/adapters/postgresadmin/testpostgres_test.go b/internal/adapters/postgresadmin/testpostgres_test.go new file mode 100644 index 0000000..7d678ba --- /dev/null +++ b/internal/adapters/postgresadmin/testpostgres_test.go @@ -0,0 +1,98 @@ +//go:build integration + +package postgresadmin + +import ( + "context" + "fmt" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "proxy-pool/internal/domain/adminstate" +) + +var postgresTestSchemaSequence atomic.Uint64 + +type postgresTestFixture struct { + Store adminstate.Store + Pool *pgxpool.Pool + Schema string + Cleanup func() +} + +func newPostgresTestFixture(t *testing.T) postgresTestFixture { + t.Helper() + postgresURL := os.Getenv("PROXY_POOL_TEST_POSTGRES_URL") + if postgresURL == "" { + t.Skip("PROXY_POOL_TEST_POSTGRES_URL is not set") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + adminConfig, err := pgxpool.ParseConfig(postgresURL) + if err != nil { + t.Fatalf("parse PROXY_POOL_TEST_POSTGRES_URL: %v", err) + } + adminPool, err := pgxpool.NewWithConfig(ctx, adminConfig) + if err != nil { + t.Fatalf("connect test PostgreSQL: %v", err) + } + if err := adminPool.Ping(ctx); err != nil { + adminPool.Close() + t.Fatalf("ping test PostgreSQL: %v", err) + } + + schema := fmt.Sprintf("it_%d_%d_%d", os.Getpid(), time.Now().UnixNano(), postgresTestSchemaSequence.Add(1)) + quotedSchema := pgx.Identifier{schema}.Sanitize() + if _, err := adminPool.Exec(ctx, "CREATE SCHEMA "+quotedSchema); err != nil { + adminPool.Close() + t.Fatalf("create isolated PostgreSQL schema: %v", err) + } + + testConfig, err := pgxpool.ParseConfig(postgresURL) + if err != nil { + adminPool.Close() + t.Fatalf("parse isolated PostgreSQL config: %v", err) + } + testConfig.ConnConfig.RuntimeParams["search_path"] = schema + testPool, err := pgxpool.NewWithConfig(ctx, testConfig) + if err != nil { + _, _ = adminPool.Exec(context.Background(), "DROP SCHEMA "+quotedSchema+" CASCADE") + adminPool.Close() + t.Fatalf("connect isolated PostgreSQL schema: %v", err) + } + if err := ApplyMigrations(ctx, testPool); err != nil { + testPool.Close() + _, _ = adminPool.Exec(context.Background(), "DROP SCHEMA "+quotedSchema+" CASCADE") + adminPool.Close() + t.Fatalf("apply PostgreSQL migrations: %v", err) + } + store, err := New(testPool) + if err != nil { + testPool.Close() + _, _ = adminPool.Exec(context.Background(), "DROP SCHEMA "+quotedSchema+" CASCADE") + adminPool.Close() + t.Fatalf("construct PostgreSQL adapter: %v", err) + } + + var cleanupOnce sync.Once + cleanup := func() { + cleanupOnce.Do(func() { + testPool.Close() + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cleanupCancel() + if _, err := adminPool.Exec(cleanupCtx, "DROP SCHEMA "+quotedSchema+" CASCADE"); err != nil { + t.Errorf("drop isolated PostgreSQL schema: %v", err) + } + adminPool.Close() + }) + } + t.Cleanup(cleanup) + return postgresTestFixture{Store: store, Pool: testPool, Schema: schema, Cleanup: cleanup} +} diff --git a/scripts/test-postgres.ps1 b/scripts/test-postgres.ps1 new file mode 100644 index 0000000..d4f515e --- /dev/null +++ b/scripts/test-postgres.ps1 @@ -0,0 +1,34 @@ +$ErrorActionPreference = "Stop" + +$repositoryRoot = Split-Path -Parent $PSScriptRoot +$composeFile = Join-Path $repositoryRoot "deploy/docker-compose.test.yml" +$composeProject = "proxy-pool-postgres-test" +$previousPostgresURL = [Environment]::GetEnvironmentVariable("PROXY_POOL_TEST_POSTGRES_URL", "Process") + +try { + docker compose -p $composeProject -f $composeFile up -d --wait --wait-timeout 60 postgres + if ($LASTEXITCODE -ne 0) { + throw "starting PostgreSQL test fixture failed with exit code $LASTEXITCODE" + } + + $env:PROXY_POOL_TEST_POSTGRES_URL = "postgres://proxy_pool_test:proxy-pool-test@127.0.0.1:15432/proxy_pool_test?sslmode=disable" + Push-Location $repositoryRoot + try { + go test -count=1 -tags=integration -timeout 60s ./internal/adapters/postgresadmin/... + if ($LASTEXITCODE -ne 0) { + throw "PostgreSQL integration tests failed with exit code $LASTEXITCODE" + } + } + finally { + Pop-Location + } +} +finally { + if ($null -eq $previousPostgresURL) { + Remove-Item Env:PROXY_POOL_TEST_POSTGRES_URL -ErrorAction SilentlyContinue + } + else { + $env:PROXY_POOL_TEST_POSTGRES_URL = $previousPostgresURL + } + docker compose -p $composeProject -f $composeFile down --volumes --remove-orphans +}