From 86e3522069dc1cbe20ac08395109d2e9e60380b5 Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Fri, 31 Jul 2026 17:26:42 +0200 Subject: [PATCH 1/7] chore: pyt redirects contract tests --- .../pytransformer_contract/redirects_test.go | 434 ++++++++++++++++++ 1 file changed, 434 insertions(+) create mode 100644 integration_test/pytransformer_contract/redirects_test.go diff --git a/integration_test/pytransformer_contract/redirects_test.go b/integration_test/pytransformer_contract/redirects_test.go new file mode 100644 index 000000000..464c8be64 --- /dev/null +++ b/integration_test/pytransformer_contract/redirects_test.go @@ -0,0 +1,434 @@ +package pytransformer_contract + +import ( + "bytes" + "context" + "database/sql" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/ory/dockertest/v3" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" + + "github.com/rudderlabs/rudder-go-kit/config" + "github.com/rudderlabs/rudder-go-kit/jsonrs" + kithelper "github.com/rudderlabs/rudder-go-kit/testhelper" + "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres" + + "github.com/rudderlabs/rudder-server/jobsdb" + "github.com/rudderlabs/rudder-server/processor/types" + "github.com/rudderlabs/rudder-server/testhelper/backendconfigtest" + "github.com/rudderlabs/rudder-server/testhelper/health" + "github.com/rudderlabs/rudder-server/testhelper/transformertest" + reportingtypes "github.com/rudderlabs/rudder-server/utils/types" +) + +// TestConfigBackendRedirectResponse pins what pytransformer puts on the wire when the config +// backend answers a transformation-code fetch with a 3xx — a redirecting proxy or gateway +// sitting in front of it. +// +// pytransformer does not follow redirects, so the 3xx reaches its response_status_handler and +// raises ConfigBackendRedirectError: HTTP 503 plus the retry headers, the same treatment as a +// 401/403, because a redirect is our misconfiguration and not the customer's code. +// +// Deliberately a raw POST rather than usertransformer.Client: the client retries 503 + +// X-Rudder-Should-Retry indefinitely by design, which is exactly the behaviour +// TestConfigBackendRedirectIsRetriedNotDropped depends on and exactly what would hang here. +func TestConfigBackendRedirectResponse(t *testing.T) { + const versionID = "redirect-test-v1" + + pythonCode := ` +def transformEvent(event, metadata): + event['foo'] = 'bar' + return event +` + + for _, redirectStatus := range []int{301, 302, 303, 307, 308} { + t.Run(fmt.Sprintf("status_%d", redirectStatus), func(t *testing.T) { + pool, err := dockertest.NewPool("") + require.NoError(t, err) + + cb := newRedirectingConfigBackend(t, redirectStatus, pythonCode) + pyTransformerURL := startRudderPytransformer(t, pool, cb.backend.URL) + + status, headers, items := postCustomTransform(t, pyTransformerURL, versionID) + + t.Logf("pytransformer returned HTTP %d, should-retry=%q reason=%q", + status, headers.Get("X-Rudder-Should-Retry"), headers.Get("X-Rudder-Error-Reason")) + for _, it := range items { + t.Logf(" item: statusCode=%d error=%q", it.StatusCode, it.Error) + } + + require.Positive(t, cb.backendHits.Load(), + "config backend was never asked for the transformation code") + require.Zero(t, cb.targetHits.Load(), + "redirect was followed: the target server received a request") + + // The whole point: retryable, so rudder-server holds the events instead of + // aborting them. A 200-level batch here is silent data loss. + require.Equal(t, http.StatusServiceUnavailable, status) + require.Equal(t, "true", headers.Get("X-Rudder-Should-Retry")) + require.Equal(t, "config_backend_redirect", headers.Get("X-Rudder-Error-Reason")) + + require.Len(t, items, 1) + require.Equal(t, http.StatusServiceUnavailable, items[0].StatusCode) + // The message must name the redirect. The old one said "Transformation not + // found", which sent operators hunting for a bad versionId. + require.Contains(t, items[0].Error, "redirect") + }) + } +} + +// TestConfigBackendRedirectIsRetriedNotDropped is the regression test for the data loss. +// +// Before ConfigBackendRedirectError existed, a 3xx was terminal, and this same scenario +// produced `pu=user_transformer status=aborted status_code=302 count=5` with zero events +// delivered and no dead-letter table to replay them from. One misconfigured proxy destroyed +// every event for every Python transformation, while JS transformations sailed through. +// +// So it proves the events survive, in the only way that really counts: hold the config +// backend broken for a while and show nothing is aborted, then repair it and show the +// original events come out the far end transformed. Recovery is the assertion — "not +// aborted yet" alone would pass on a pipeline that was merely slow. +func TestConfigBackendRedirectIsRetriedNotDropped(t *testing.T) { + const ( + redirectStatus = http.StatusFound // 302 + eventsCount = 5 + ) + + pythonCode := ` +def transformEvent(event, metadata): + event['foo'] = 'bar' + return event +` + + pool, err := dockertest.NewPool("") + require.NoError(t, err) + + postgresContainer, err := postgres.Setup(pool, t) + require.NoError(t, err) + + webhookServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer webhookServer.Close() + + cb := newRedirectingConfigBackend(t, redirectStatus, pythonCode) + pyTransformerURL := startRudderPytransformer(t, pool, cb.backend.URL) + + trServer := transformertest.NewBuilder().Build() + defer trServer.Close() + + bcServer := backendconfigtest.NewBuilder(). + WithWorkspaceConfig( + backendconfigtest.NewConfigBuilder(). + WithSource( + backendconfigtest.NewSourceBuilder(). + WithID("source-1"). + WithWriteKey("writekey-1"). + WithConnection( + backendconfigtest.NewDestinationBuilder("WEBHOOK"). + WithID("destination-1"). + WithUserTransformation("transformation-1", "version-1"). + WithConfigOption("webhookUrl", webhookServer.URL). + Build()). + Build()). + Build()). + Build() + defer bcServer.Close() + + // Keep the retry backoff short so the pipeline recovers promptly once the config + // backend is repaired. Retries stay unbounded — that is the behaviour under test. + t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, + "Transformer.Client.UserTransformer.retryRudderErrors.maxInterval"), "100ms") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + gwPort, err := kithelper.GetFreePort() + require.NoError(t, err) + + wg, ctx := errgroup.WithContext(ctx) + wg.Go(func() error { + err := runRudderServer( + t, ctx, cancel, gwPort, postgresContainer, bcServer.URL, trServer.URL, pyTransformerURL, t.TempDir(), + ) + if err != nil { + t.Logf("rudder-server exited with error: %v", err) + } + return err + }) + + url := fmt.Sprintf("http://localhost:%d", gwPort) + health.WaitUntilReady(ctx, t, url+"/health", 60*time.Second, 10*time.Millisecond, t.Name()) + + t.Logf("Sending %d identify events while the config backend redirects...", eventsCount) + require.NoError(t, sendEvents(eventsCount, "identify", "writekey-1", url)) + + // Phase 1 — broken. Nothing may be aborted and nothing may be delivered. + // + // Note the gateway jobs are deliberately NOT asserted succeeded here: the processor is + // blocked retrying the fetch, so they stay in flight. That they are still in flight + // rather than resolved is the point — under the old terminal behaviour they would + // already have been marked done, with the events gone. + t.Log("--- phase 1: config backend redirecting ---") + hitsBefore := cb.backendHits.Load() + requireNoAbortedUserTransformations(t, ctx, postgresContainer.DB) + logAllJobsTableCounts(t, ctx, postgresContainer.DB) + require.Greater(t, cb.backendHits.Load(), hitsBefore, + "pytransformer should still be retrying the fetch, so the config backend keeps being hit") + + // Phase 2 — repaired. The same events must come out transformed. + t.Log("--- phase 2: config backend repaired ---") + cb.serveCode() + + requireJobsCount(t, ctx, postgresContainer.DB, "gw", jobsdb.Succeeded.State, eventsCount) + requireJobsCount(t, ctx, postgresContainer.DB, "rt", jobsdb.Succeeded.State, eventsCount) + requireTransformationApplied(t, ctx, postgresContainer.DB, eventsCount) + logAllJobsTableCounts(t, ctx, postgresContainer.DB) + logReportedStatuses(t, ctx, postgresContainer.DB) + + require.Zero(t, cb.targetHits.Load(), + "redirect was followed: the target server received a request") + + t.Logf("RESULT: all %d events survived the misconfiguration and were delivered transformed", + eventsCount) + + cancel() + require.NoError(t, wg.Wait()) +} + +// redirectingConfigBackend is a config backend that answers /transformation/getByVersionId +// with a redirect until serveCode is called, after which it serves the transformation +// normally. The flip is what lets a test show that events were held rather than destroyed. +// +// It counts hits on both ends. The target counter is the load-bearing one: it distinguishes +// "the redirect was refused" from "the redirect was followed", which changes what the test is +// actually observing. The Location is built with toContainerURL so the target is genuinely +// reachable from inside the pytransformer container — otherwise a followed redirect would +// surface as a connection error and look like a refusal. +type redirectingConfigBackend struct { + backend *httptest.Server + target *httptest.Server + backendHits atomic.Int64 + targetHits atomic.Int64 + redirecting atomic.Bool + redirectStatus int + code string +} + +func newRedirectingConfigBackend(t *testing.T, redirectStatus int, code string) *redirectingConfigBackend { + t.Helper() + + cb := &redirectingConfigBackend{redirectStatus: redirectStatus, code: code} + cb.redirecting.Store(true) + + // Where the redirect points. Serves a perfectly valid transformation, so if the + // redirect were followed the fetch would succeed and the test would fail loudly on + // targetHits rather than silently passing for the wrong reason. + cb.target = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cb.targetHits.Add(1) + t.Logf("REDIRECT TARGET was contacted: %s %s", r.Method, r.URL.Path) + cb.writeCode(w) + })) + t.Cleanup(cb.target.Close) + + cb.backend = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/transformation/getByVersionId", "/transformationLibrary/getByVersionId": + cb.backendHits.Add(1) + if !cb.redirecting.Load() { + cb.writeCode(w) + return + } + location := toContainerURL(cb.target.URL) + r.URL.Path + "?" + r.URL.RawQuery + w.Header().Set("Location", location) + w.WriteHeader(cb.redirectStatus) + default: + t.Logf("CONFIG BACKEND: unexpected path %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(cb.backend.Close) + + return cb +} + +// serveCode repairs the config backend: subsequent fetches get the transformation. +func (cb *redirectingConfigBackend) serveCode() { + cb.redirecting.Store(false) +} + +func (cb *redirectingConfigBackend) writeCode(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"code":%q,"language":"pythonfaas","codeVersion":"1"}`, cb.code) +} + +// postCustomTransform posts a single event straight to pytransformer's /customTransform and +// returns the raw status, headers and decoded items. sendRawTransform in +// dns_cache_contract_test.go does almost this, but discards the headers, and the retry +// headers are the contract under test here. +func postCustomTransform( + t *testing.T, + baseURL, versionID string, +) (int, http.Header, []types.TransformerResponse) { + t.Helper() + + body, err := jsonrs.Marshal([]any{ + map[string]any{ + "message": map[string]any{"messageId": "msg-1", "type": "track", "event": "Test Event"}, + "metadata": map[string]any{ + "sourceId": "src-1", "destinationId": "dest-1", + "workspaceId": "ws-1", "messageId": "msg-1", + }, + "destination": map[string]any{ + "Transformations": []any{ + map[string]any{"VersionID": versionID, "ID": "transformation-1", "Language": "pythonfaas"}, + }, + }, + }, + }) + require.NoError(t, err) + + req, err := http.NewRequest(http.MethodPost, baseURL+"/customTransform", bytes.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + + var items []types.TransformerResponse + require.NoError(t, jsonrs.NewDecoder(resp.Body).Decode(&items)) + + return resp.StatusCode, resp.Header.Clone(), items +} + +// requireNoAbortedUserTransformations holds for a few seconds and fails the moment anything +// is aborted at the user transformation stage or queued for delivery. require.Never rather +// than a point-in-time check, so a pipeline that has simply not got there yet cannot pass. +func requireNoAbortedUserTransformations(t *testing.T, ctx context.Context, db *sql.DB) { + t.Helper() + + require.Never(t, func() bool { + var aborted int + err := db.QueryRowContext(ctx, ` + SELECT coalesce(sum(count), 0) FROM reports + WHERE pu = $1 AND status = 'aborted' + `, reportingtypes.USER_TRANSFORMER).Scan(&aborted) + if err == nil && aborted > 0 { + t.Logf("ABORTED: %d events aborted at the user transformation stage", aborted) + return true + } + + for _, table := range []string{"rt_jobs_1", "batch_rt_jobs_1"} { + var count int + //nolint:gosec // fixed table names + if err := db.QueryRowContext(ctx, "SELECT count(*) FROM "+table).Scan(&count); err != nil { + continue // a missing table is itself proof nothing was queued there + } + if count > 0 { + t.Logf("DELIVERED: %s unexpectedly has %d jobs", table, count) + return true + } + } + return false + }, + 10*time.Second, + 500*time.Millisecond, + "events must be held for retry, not aborted or delivered, while the config backend redirects", + ) +} + +// logAllJobsTableCounts prints the row count of every jobs table in the database. It asserts +// nothing: it exists so the reader can see for themselves where the events are. +func logAllJobsTableCounts(t *testing.T, ctx context.Context, db *sql.DB) { + t.Helper() + + rows, err := db.QueryContext(ctx, ` + SELECT table_name + FROM information_schema.tables + WHERE table_schema = 'public' AND table_name LIKE '%_jobs_%' + ORDER BY table_name + `) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + var tables []string + for rows.Next() { + var name string + require.NoError(t, rows.Scan(&name)) + tables = append(tables, name) + } + require.NoError(t, rows.Err()) + + t.Log("--- jobs table row counts ---") + for _, table := range tables { + var count int + //nolint:gosec // table name comes from information_schema, not user input + if err := db.QueryRowContext(ctx, "SELECT count(*) FROM "+table).Scan(&count); err != nil { + t.Logf(" %-24s ", table, err) + continue + } + t.Logf(" %-24s %d", table, count) + } + t.Log("-----------------------------") +} + +// logReportedStatuses prints what the reporting module recorded per pipeline stage. Evidence +// only — reporting is not guaranteed to be on in this setup, so nothing is asserted. +func logReportedStatuses(t *testing.T, ctx context.Context, db *sql.DB) { + t.Helper() + + var exists bool + err := db.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'reports' + ) + `).Scan(&exists) + if err != nil || !exists { + t.Logf("no reports table to inspect (err=%v, exists=%v)", err, exists) + return + } + + rows, err := db.QueryContext(ctx, ` + SELECT pu, status, status_code, count, coalesce(sample_response, '') + FROM reports + ORDER BY pu + `) + if err != nil { + t.Logf("querying reports: %v", err) + return + } + defer func() { _ = rows.Close() }() + + t.Log("--- reported statuses ---") + var any bool + for rows.Next() { + var pu, status, sample string + var statusCode, count int + if err := rows.Scan(&pu, &status, &statusCode, &count, &sample); err != nil { + t.Logf("scanning reports: %v", err) + return + } + any = true + marker := "" + if pu == reportingtypes.USER_TRANSFORMER { + marker = " <-- user transformation stage" + } + t.Logf(" pu=%-18s status=%-10s status_code=%d count=%d%s", pu, status, statusCode, count, marker) + if sample != "" { + t.Logf(" sample: %s", sample) + } + } + if !any { + t.Log(" (no rows)") + } + t.Log("-------------------------") +} From 031a6c0457b41860eb3a17e7c9747897196cdaad Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Fri, 31 Jul 2026 17:47:38 +0200 Subject: [PATCH 2/7] chore: simplifying tests --- .../dns_cache_contract_test.go | 18 +- .../pytransformer_contract/redirects_test.go | 269 +++++------------- 2 files changed, 86 insertions(+), 201 deletions(-) diff --git a/integration_test/pytransformer_contract/dns_cache_contract_test.go b/integration_test/pytransformer_contract/dns_cache_contract_test.go index 06db37970..f2fa85e61 100644 --- a/integration_test/pytransformer_contract/dns_cache_contract_test.go +++ b/integration_test/pytransformer_contract/dns_cache_contract_test.go @@ -445,6 +445,22 @@ func sendRawTransform( ) ( int, []types.TransformerResponse, +) { + t.Helper() + status, _, items := sendRawTransformWithHeaders(t, baseURL, events) + return status, items +} + +// sendRawTransformWithHeaders is sendRawTransform plus the response headers, for tests that +// assert the retry contract (X-Rudder-Should-Retry / X-Rudder-Error-Reason). +func sendRawTransformWithHeaders( + t *testing.T, + baseURL string, + events []types.TransformerEvent, +) ( + int, + http.Header, + []types.TransformerResponse, ) { t.Helper() payload := make([]any, len(events)) @@ -471,5 +487,5 @@ func sendRawTransform( var items []types.TransformerResponse require.NoError(t, jsonrs.NewDecoder(resp.Body).Decode(&items)) - return resp.StatusCode, items + return resp.StatusCode, resp.Header.Clone(), items } diff --git a/integration_test/pytransformer_contract/redirects_test.go b/integration_test/pytransformer_contract/redirects_test.go index 464c8be64..e12f2616b 100644 --- a/integration_test/pytransformer_contract/redirects_test.go +++ b/integration_test/pytransformer_contract/redirects_test.go @@ -1,7 +1,6 @@ package pytransformer_contract import ( - "bytes" "context" "database/sql" "fmt" @@ -16,10 +15,10 @@ import ( "golang.org/x/sync/errgroup" "github.com/rudderlabs/rudder-go-kit/config" - "github.com/rudderlabs/rudder-go-kit/jsonrs" kithelper "github.com/rudderlabs/rudder-go-kit/testhelper" "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres" + backendconfig "github.com/rudderlabs/rudder-server/backend-config" "github.com/rudderlabs/rudder-server/jobsdb" "github.com/rudderlabs/rudder-server/processor/types" "github.com/rudderlabs/rudder-server/testhelper/backendconfigtest" @@ -42,27 +41,18 @@ import ( func TestConfigBackendRedirectResponse(t *testing.T) { const versionID = "redirect-test-v1" - pythonCode := ` -def transformEvent(event, metadata): - event['foo'] = 'bar' - return event -` - for _, redirectStatus := range []int{301, 302, 303, 307, 308} { t.Run(fmt.Sprintf("status_%d", redirectStatus), func(t *testing.T) { pool, err := dockertest.NewPool("") require.NoError(t, err) - cb := newRedirectingConfigBackend(t, redirectStatus, pythonCode) - pyTransformerURL := startRudderPytransformer(t, pool, cb.backend.URL) + cb := newRedirectingConfigBackend(t, redirectStatus) + pyURL := startRudderPytransformer(t, pool, cb.backend.URL) - status, headers, items := postCustomTransform(t, pyTransformerURL, versionID) + status, headers, items := sendRawTransformWithHeaders(t, pyURL, redirectTestEvents(versionID, 1)) t.Logf("pytransformer returned HTTP %d, should-retry=%q reason=%q", status, headers.Get("X-Rudder-Should-Retry"), headers.Get("X-Rudder-Error-Reason")) - for _, it := range items { - t.Logf(" item: statusCode=%d error=%q", it.StatusCode, it.Error) - } require.Positive(t, cb.backendHits.Load(), "config backend was never asked for the transformation code") @@ -91,22 +81,16 @@ def transformEvent(event, metadata): // delivered and no dead-letter table to replay them from. One misconfigured proxy destroyed // every event for every Python transformation, while JS transformations sailed through. // -// So it proves the events survive, in the only way that really counts: hold the config -// backend broken for a while and show nothing is aborted, then repair it and show the -// original events come out the far end transformed. Recovery is the assertion — "not -// aborted yet" alone would pass on a pipeline that was merely slow. +// The load-bearing assertion is the recovery: hold the config backend broken, then repair it +// and show the *original* events come out the far end transformed. Aborted events would be +// gone, so phase 2 could not pass. The aborted-count check in phase 1 is there to fail early +// with a clear message rather than as a 60s timeout in phase 2. func TestConfigBackendRedirectIsRetriedNotDropped(t *testing.T) { const ( redirectStatus = http.StatusFound // 302 eventsCount = 5 ) - pythonCode := ` -def transformEvent(event, metadata): - event['foo'] = 'bar' - return event -` - pool, err := dockertest.NewPool("") require.NoError(t, err) @@ -118,7 +102,7 @@ def transformEvent(event, metadata): })) defer webhookServer.Close() - cb := newRedirectingConfigBackend(t, redirectStatus, pythonCode) + cb := newRedirectingConfigBackend(t, redirectStatus) pyTransformerURL := startRudderPytransformer(t, pool, cb.backend.URL) trServer := transformertest.NewBuilder().Build() @@ -170,16 +154,15 @@ def transformEvent(event, metadata): t.Logf("Sending %d identify events while the config backend redirects...", eventsCount) require.NoError(t, sendEvents(eventsCount, "identify", "writekey-1", url)) - // Phase 1 — broken. Nothing may be aborted and nothing may be delivered. + // Phase 1 — broken. Nothing may be aborted. // - // Note the gateway jobs are deliberately NOT asserted succeeded here: the processor is + // The gateway jobs are deliberately NOT asserted succeeded here: the processor is // blocked retrying the fetch, so they stay in flight. That they are still in flight // rather than resolved is the point — under the old terminal behaviour they would // already have been marked done, with the events gone. t.Log("--- phase 1: config backend redirecting ---") hitsBefore := cb.backendHits.Load() requireNoAbortedUserTransformations(t, ctx, postgresContainer.DB) - logAllJobsTableCounts(t, ctx, postgresContainer.DB) require.Greater(t, cb.backendHits.Load(), hitsBefore, "pytransformer should still be retrying the fetch, so the config backend keeps being hit") @@ -190,8 +173,6 @@ def transformEvent(event, metadata): requireJobsCount(t, ctx, postgresContainer.DB, "gw", jobsdb.Succeeded.State, eventsCount) requireJobsCount(t, ctx, postgresContainer.DB, "rt", jobsdb.Succeeded.State, eventsCount) requireTransformationApplied(t, ctx, postgresContainer.DB, eventsCount) - logAllJobsTableCounts(t, ctx, postgresContainer.DB) - logReportedStatuses(t, ctx, postgresContainer.DB) require.Zero(t, cb.targetHits.Load(), "redirect was followed: the target server received a request") @@ -203,38 +184,70 @@ def transformEvent(event, metadata): require.NoError(t, wg.Wait()) } +// redirectTestEvents builds n events bound to versionID, in the shape sendRawTransform wants. +func redirectTestEvents(versionID string, n int) []types.TransformerEvent { + events := make([]types.TransformerEvent, n) + for i := range events { + messageID := fmt.Sprintf("msg-%d", i+1) + events[i] = types.TransformerEvent{ + Message: types.SingularEventT{ + "messageId": messageID, "type": "track", "event": "Test Event", + }, + Metadata: types.Metadata{ + SourceID: "src-1", DestinationID: "dest-1", + WorkspaceID: "ws-1", MessageID: messageID, + }, + Destination: backendconfig.DestinationT{ + Transformations: []backendconfig.TransformationT{ + {VersionID: versionID, ID: "transformation-1", Language: "pythonfaas"}, + }, + }, + } + } + return events +} + // redirectingConfigBackend is a config backend that answers /transformation/getByVersionId // with a redirect until serveCode is called, after which it serves the transformation // normally. The flip is what lets a test show that events were held rather than destroyed. // -// It counts hits on both ends. The target counter is the load-bearing one: it distinguishes -// "the redirect was refused" from "the redirect was followed", which changes what the test is -// actually observing. The Location is built with toContainerURL so the target is genuinely -// reachable from inside the pytransformer container — otherwise a followed redirect would -// surface as a connection error and look like a refusal. +// newContractConfigBackend can return an arbitrary status, but not a Location header and not +// a mid-test flip, so this is its own fixture. It counts hits on both ends: the target +// counter is the load-bearing one, distinguishing "the redirect was refused" from "the +// redirect was followed", which changes what the test is actually observing. The Location is +// built with toContainerURL so the target is genuinely reachable from inside the pytransformer +// container — otherwise a followed redirect would surface as a connection error and look like +// a refusal. type redirectingConfigBackend struct { - backend *httptest.Server - target *httptest.Server - backendHits atomic.Int64 - targetHits atomic.Int64 - redirecting atomic.Bool - redirectStatus int - code string + backend *httptest.Server + target *httptest.Server + backendHits atomic.Int64 + targetHits atomic.Int64 + redirecting atomic.Bool } -func newRedirectingConfigBackend(t *testing.T, redirectStatus int, code string) *redirectingConfigBackend { +func newRedirectingConfigBackend(t *testing.T, redirectStatus int) *redirectingConfigBackend { t.Helper() - cb := &redirectingConfigBackend{redirectStatus: redirectStatus, code: code} + cb := &redirectingConfigBackend{} cb.redirecting.Store(true) + writeCode := func(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"code":%q,"language":"pythonfaas","codeVersion":"1"}`, ` +def transformEvent(event, metadata): + event['foo'] = 'bar' + return event +`) + } + // Where the redirect points. Serves a perfectly valid transformation, so if the // redirect were followed the fetch would succeed and the test would fail loudly on // targetHits rather than silently passing for the wrong reason. cb.target = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { cb.targetHits.Add(1) t.Logf("REDIRECT TARGET was contacted: %s %s", r.Method, r.URL.Path) - cb.writeCode(w) + writeCode(w) })) t.Cleanup(cb.target.Close) @@ -243,12 +256,11 @@ func newRedirectingConfigBackend(t *testing.T, redirectStatus int, code string) case "/transformation/getByVersionId", "/transformationLibrary/getByVersionId": cb.backendHits.Add(1) if !cb.redirecting.Load() { - cb.writeCode(w) + writeCode(w) return } - location := toContainerURL(cb.target.URL) + r.URL.Path + "?" + r.URL.RawQuery - w.Header().Set("Location", location) - w.WriteHeader(cb.redirectStatus) + w.Header().Set("Location", toContainerURL(cb.target.URL)+r.URL.Path+"?"+r.URL.RawQuery) + w.WriteHeader(redirectStatus) default: t.Logf("CONFIG BACKEND: unexpected path %s", r.URL.Path) w.WriteHeader(http.StatusNotFound) @@ -264,54 +276,9 @@ func (cb *redirectingConfigBackend) serveCode() { cb.redirecting.Store(false) } -func (cb *redirectingConfigBackend) writeCode(w http.ResponseWriter) { - w.Header().Set("Content-Type", "application/json") - _, _ = fmt.Fprintf(w, `{"code":%q,"language":"pythonfaas","codeVersion":"1"}`, cb.code) -} - -// postCustomTransform posts a single event straight to pytransformer's /customTransform and -// returns the raw status, headers and decoded items. sendRawTransform in -// dns_cache_contract_test.go does almost this, but discards the headers, and the retry -// headers are the contract under test here. -func postCustomTransform( - t *testing.T, - baseURL, versionID string, -) (int, http.Header, []types.TransformerResponse) { - t.Helper() - - body, err := jsonrs.Marshal([]any{ - map[string]any{ - "message": map[string]any{"messageId": "msg-1", "type": "track", "event": "Test Event"}, - "metadata": map[string]any{ - "sourceId": "src-1", "destinationId": "dest-1", - "workspaceId": "ws-1", "messageId": "msg-1", - }, - "destination": map[string]any{ - "Transformations": []any{ - map[string]any{"VersionID": versionID, "ID": "transformation-1", "Language": "pythonfaas"}, - }, - }, - }, - }) - require.NoError(t, err) - - req, err := http.NewRequest(http.MethodPost, baseURL+"/customTransform", bytes.NewReader(body)) - require.NoError(t, err) - req.Header.Set("Content-Type", "application/json") - - resp, err := http.DefaultClient.Do(req) - require.NoError(t, err) - defer func() { _ = resp.Body.Close() }() - - var items []types.TransformerResponse - require.NoError(t, jsonrs.NewDecoder(resp.Body).Decode(&items)) - - return resp.StatusCode, resp.Header.Clone(), items -} - // requireNoAbortedUserTransformations holds for a few seconds and fails the moment anything -// is aborted at the user transformation stage or queued for delivery. require.Never rather -// than a point-in-time check, so a pipeline that has simply not got there yet cannot pass. +// is aborted at the user transformation stage. require.Never rather than a point-in-time +// check, so a pipeline that has simply not got there yet cannot pass. func requireNoAbortedUserTransformations(t *testing.T, ctx context.Context, db *sql.DB) { t.Helper() @@ -321,114 +288,16 @@ func requireNoAbortedUserTransformations(t *testing.T, ctx context.Context, db * SELECT coalesce(sum(count), 0) FROM reports WHERE pu = $1 AND status = 'aborted' `, reportingtypes.USER_TRANSFORMER).Scan(&aborted) - if err == nil && aborted > 0 { - t.Logf("ABORTED: %d events aborted at the user transformation stage", aborted) - return true + if err != nil { + return false // reporting not up yet; phase 2 is the real proof either way } - - for _, table := range []string{"rt_jobs_1", "batch_rt_jobs_1"} { - var count int - //nolint:gosec // fixed table names - if err := db.QueryRowContext(ctx, "SELECT count(*) FROM "+table).Scan(&count); err != nil { - continue // a missing table is itself proof nothing was queued there - } - if count > 0 { - t.Logf("DELIVERED: %s unexpectedly has %d jobs", table, count) - return true - } + if aborted > 0 { + t.Logf("ABORTED: %d events aborted at the user transformation stage", aborted) } - return false + return aborted > 0 }, 10*time.Second, 500*time.Millisecond, - "events must be held for retry, not aborted or delivered, while the config backend redirects", + "events must be held for retry, not aborted, while the config backend redirects", ) } - -// logAllJobsTableCounts prints the row count of every jobs table in the database. It asserts -// nothing: it exists so the reader can see for themselves where the events are. -func logAllJobsTableCounts(t *testing.T, ctx context.Context, db *sql.DB) { - t.Helper() - - rows, err := db.QueryContext(ctx, ` - SELECT table_name - FROM information_schema.tables - WHERE table_schema = 'public' AND table_name LIKE '%_jobs_%' - ORDER BY table_name - `) - require.NoError(t, err) - defer func() { _ = rows.Close() }() - - var tables []string - for rows.Next() { - var name string - require.NoError(t, rows.Scan(&name)) - tables = append(tables, name) - } - require.NoError(t, rows.Err()) - - t.Log("--- jobs table row counts ---") - for _, table := range tables { - var count int - //nolint:gosec // table name comes from information_schema, not user input - if err := db.QueryRowContext(ctx, "SELECT count(*) FROM "+table).Scan(&count); err != nil { - t.Logf(" %-24s ", table, err) - continue - } - t.Logf(" %-24s %d", table, count) - } - t.Log("-----------------------------") -} - -// logReportedStatuses prints what the reporting module recorded per pipeline stage. Evidence -// only — reporting is not guaranteed to be on in this setup, so nothing is asserted. -func logReportedStatuses(t *testing.T, ctx context.Context, db *sql.DB) { - t.Helper() - - var exists bool - err := db.QueryRowContext(ctx, ` - SELECT EXISTS ( - SELECT 1 FROM information_schema.tables - WHERE table_schema = 'public' AND table_name = 'reports' - ) - `).Scan(&exists) - if err != nil || !exists { - t.Logf("no reports table to inspect (err=%v, exists=%v)", err, exists) - return - } - - rows, err := db.QueryContext(ctx, ` - SELECT pu, status, status_code, count, coalesce(sample_response, '') - FROM reports - ORDER BY pu - `) - if err != nil { - t.Logf("querying reports: %v", err) - return - } - defer func() { _ = rows.Close() }() - - t.Log("--- reported statuses ---") - var any bool - for rows.Next() { - var pu, status, sample string - var statusCode, count int - if err := rows.Scan(&pu, &status, &statusCode, &count, &sample); err != nil { - t.Logf("scanning reports: %v", err) - return - } - any = true - marker := "" - if pu == reportingtypes.USER_TRANSFORMER { - marker = " <-- user transformation stage" - } - t.Logf(" pu=%-18s status=%-10s status_code=%d count=%d%s", pu, status, statusCode, count, marker) - if sample != "" { - t.Logf(" sample: %s", sample) - } - } - if !any { - t.Log(" (no rows)") - } - t.Log("-------------------------") -} From 61dface1f3d0fda749f8e99c34a5a4ee92584f84 Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Mon, 3 Aug 2026 09:32:24 +0200 Subject: [PATCH 3/7] chore: expected count --- .../pytransformer_contract/pytransformer_contract_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/integration_test/pytransformer_contract/pytransformer_contract_test.go b/integration_test/pytransformer_contract/pytransformer_contract_test.go index cd939194d..35d486ba4 100644 --- a/integration_test/pytransformer_contract/pytransformer_contract_test.go +++ b/integration_test/pytransformer_contract/pytransformer_contract_test.go @@ -137,7 +137,7 @@ def transformEvent(event, metadata): t.Logf("rudder-server is ready at %s", url) // 10. Send events to rudder-server - eventsCount := 5 + eventsCount := 6 t.Logf("Sending %d identify events...", eventsCount) err = sendEvents(eventsCount, "identify", "writekey-1", url) require.NoError(t, err) @@ -264,6 +264,7 @@ func sendEvents( return nil } +// requireJobsCount waits until a jobsdb queue holds expectedCount jobs in a given state. func requireJobsCount( t *testing.T, ctx context.Context, From 911d3bc167251cf6c555bd6e12dc2ea6211ad643 Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Mon, 3 Aug 2026 10:34:02 +0200 Subject: [PATCH 4/7] chore: cleaning up --- .../pytransformer_contract/bc_helpers_test.go | 18 +++++++++++++ .../dns_cache_contract_test.go | 8 +++++- .../pytransformer_contract/redirects_test.go | 27 +------------------ 3 files changed, 26 insertions(+), 27 deletions(-) diff --git a/integration_test/pytransformer_contract/bc_helpers_test.go b/integration_test/pytransformer_contract/bc_helpers_test.go index d03b2d8a9..5b06a469c 100644 --- a/integration_test/pytransformer_contract/bc_helpers_test.go +++ b/integration_test/pytransformer_contract/bc_helpers_test.go @@ -248,6 +248,24 @@ func makeEventWithCredentials(messageID, versionID string, credentials []types.C return ev } +// makeEvents creates n TransformerEvents for versionID, optionally carrying library version ids. +// +// Message ids are prefixed with versionID so a test sharing one mock config backend across +// subtests can scope its assertions to its own requests. +func makeEvents(versionID string, n int, libraryVersionIDs ...string) []types.TransformerEvent { + libraries := make([]backendconfig.LibraryT, len(libraryVersionIDs)) + for i, id := range libraryVersionIDs { + libraries[i] = backendconfig.LibraryT{VersionID: id} + } + + events := make([]types.TransformerEvent, n) + for i := range events { + events[i] = makeEvent(fmt.Sprintf("%s-msg-%d", versionID, i+1), versionID) + events[i].Libraries = libraries + } + return events +} + // configBackendEntry controls what the mock config backend returns for a given versionId. // // When statusCode is 0 (default), the entry is treated as a normal transformation: diff --git a/integration_test/pytransformer_contract/dns_cache_contract_test.go b/integration_test/pytransformer_contract/dns_cache_contract_test.go index f2fa85e61..b05963617 100644 --- a/integration_test/pytransformer_contract/dns_cache_contract_test.go +++ b/integration_test/pytransformer_contract/dns_cache_contract_test.go @@ -465,13 +465,19 @@ func sendRawTransformWithHeaders( t.Helper() payload := make([]any, len(events)) for i, ev := range events { - payload[i] = map[string]any{ + item := map[string]any{ "message": ev.Message, "metadata": ev.Metadata, "destination": map[string]any{ "Transformations": ev.Destination.Transformations, }, } + // Only when set: pytransformer reads libraries per event, and an empty array would + // change the cache key for every test that does not use libraries. + if len(ev.Libraries) > 0 { + item["libraries"] = ev.Libraries + } + payload[i] = item } body, err := jsonrs.Marshal(payload) require.NoError(t, err) diff --git a/integration_test/pytransformer_contract/redirects_test.go b/integration_test/pytransformer_contract/redirects_test.go index e12f2616b..7922d85d8 100644 --- a/integration_test/pytransformer_contract/redirects_test.go +++ b/integration_test/pytransformer_contract/redirects_test.go @@ -18,9 +18,7 @@ import ( kithelper "github.com/rudderlabs/rudder-go-kit/testhelper" "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres" - backendconfig "github.com/rudderlabs/rudder-server/backend-config" "github.com/rudderlabs/rudder-server/jobsdb" - "github.com/rudderlabs/rudder-server/processor/types" "github.com/rudderlabs/rudder-server/testhelper/backendconfigtest" "github.com/rudderlabs/rudder-server/testhelper/health" "github.com/rudderlabs/rudder-server/testhelper/transformertest" @@ -49,7 +47,7 @@ func TestConfigBackendRedirectResponse(t *testing.T) { cb := newRedirectingConfigBackend(t, redirectStatus) pyURL := startRudderPytransformer(t, pool, cb.backend.URL) - status, headers, items := sendRawTransformWithHeaders(t, pyURL, redirectTestEvents(versionID, 1)) + status, headers, items := sendRawTransformWithHeaders(t, pyURL, makeEvents(versionID, 1)) t.Logf("pytransformer returned HTTP %d, should-retry=%q reason=%q", status, headers.Get("X-Rudder-Should-Retry"), headers.Get("X-Rudder-Error-Reason")) @@ -184,29 +182,6 @@ func TestConfigBackendRedirectIsRetriedNotDropped(t *testing.T) { require.NoError(t, wg.Wait()) } -// redirectTestEvents builds n events bound to versionID, in the shape sendRawTransform wants. -func redirectTestEvents(versionID string, n int) []types.TransformerEvent { - events := make([]types.TransformerEvent, n) - for i := range events { - messageID := fmt.Sprintf("msg-%d", i+1) - events[i] = types.TransformerEvent{ - Message: types.SingularEventT{ - "messageId": messageID, "type": "track", "event": "Test Event", - }, - Metadata: types.Metadata{ - SourceID: "src-1", DestinationID: "dest-1", - WorkspaceID: "ws-1", MessageID: messageID, - }, - Destination: backendconfig.DestinationT{ - Transformations: []backendconfig.TransformationT{ - {VersionID: versionID, ID: "transformation-1", Language: "pythonfaas"}, - }, - }, - } - } - return events -} - // redirectingConfigBackend is a config backend that answers /transformation/getByVersionId // with a redirect until serveCode is called, after which it serves the transformation // normally. The flip is what lets a test show that events were held rather than destroyed. From 71f98f7d8d0ea2b1d2fbd8a259d964e2ef971f60 Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Mon, 3 Aug 2026 10:38:34 +0200 Subject: [PATCH 5/7] chore: removing helper --- .../cookie_isolation_contract_test.go | 2 +- .../dns_cache_contract_test.go | 43 +++++++------------ .../force_thread_pool_always_contract_test.go | 4 +- .../http_timeout_conn_pool_contract_test.go | 14 +++--- .../io_bound_flag_cache_contract_test.go | 10 ++--- .../managed_session_contract_test.go | 24 +++++------ .../pytransformer_contract/redirects_test.go | 2 +- .../requests_api_contract_test.go | 4 +- .../threaded_execution_contract_test.go | 4 +- 9 files changed, 47 insertions(+), 60 deletions(-) diff --git a/integration_test/pytransformer_contract/cookie_isolation_contract_test.go b/integration_test/pytransformer_contract/cookie_isolation_contract_test.go index 60b93649c..b6008e01f 100644 --- a/integration_test/pytransformer_contract/cookie_isolation_contract_test.go +++ b/integration_test/pytransformer_contract/cookie_isolation_contract_test.go @@ -111,7 +111,7 @@ def transformEvent(event, metadata): wg.Go(func() { msgID := fmt.Sprintf("msg-cookie-iso-%d", i) events := []types.TransformerEvent{makeEvent(msgID, versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) res := result{idx: i, status: status, items: items} if len(items) == 1 && items[0].StatusCode == http.StatusOK { diff --git a/integration_test/pytransformer_contract/dns_cache_contract_test.go b/integration_test/pytransformer_contract/dns_cache_contract_test.go index b05963617..e0fd45105 100644 --- a/integration_test/pytransformer_contract/dns_cache_contract_test.go +++ b/integration_test/pytransformer_contract/dns_cache_contract_test.go @@ -103,7 +103,7 @@ def transformEvent(event, metadata): // Send one request per mock server, sequentially for _, name := range names { events := []types.TransformerEvent{makeEvent("msg-"+name, versionIDs[name])} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) requireCorrectServer(t, name, status, items) } @@ -116,7 +116,7 @@ def transformEvent(event, metadata): // Now repeat the first two — DNS cache should still resolve correctly for _, name := range []string{"alpha", "bravo"} { events := []types.TransformerEvent{makeEvent("msg-"+name+"-repeat", versionIDs[name])} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) requireCorrectServer(t, name, status, items) } @@ -144,7 +144,7 @@ def transformEvent(event, metadata): idx, n := i, name wg.Go(func() { events := []types.TransformerEvent{makeEvent("msg-parallel-"+n, versionIDs[n])} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) results[idx] = result{name: n, status: status, items: items} }) } @@ -168,7 +168,7 @@ def transformEvent(event, metadata): // Same transformation 3 times — DNS cache must remain correct for i := range 3 { events := []types.TransformerEvent{makeEvent(fmt.Sprintf("msg-repeat-%d", i), versionIDs["alpha"])} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) requireCorrectServer(t, "alpha", status, items) } @@ -224,7 +224,7 @@ def transformEvent(event, metadata): t.Run("OverrideResolvesToCorrectServer", func(t *testing.T) { events := []types.TransformerEvent{makeEvent("msg-override-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) @@ -242,7 +242,7 @@ def transformEvent(event, metadata): callsBefore := mockCalls.Load() for i := range 3 { events := []types.TransformerEvent{makeEvent(fmt.Sprintf("msg-override-repeat-%d", i), versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) @@ -325,22 +325,22 @@ def transformEvent(event, metadata): wg.Go(func() { events := []types.TransformerEvent{makeEvent("msg-combined-override-1", versionOverride)} - s, items := sendRawTransform(t, pyURL, events) + s, _, items := sendRawTransform(t, pyURL, events) results[0] = result{"override-1", "override-server", s, items} }) wg.Go(func() { events := []types.TransformerEvent{makeEvent("msg-combined-cached-1", versionCached)} - s, items := sendRawTransform(t, pyURL, events) + s, _, items := sendRawTransform(t, pyURL, events) results[1] = result{"cached-1", "cached-server", s, items} }) wg.Go(func() { events := []types.TransformerEvent{makeEvent("msg-combined-override-2", versionOverride)} - s, items := sendRawTransform(t, pyURL, events) + s, _, items := sendRawTransform(t, pyURL, events) results[2] = result{"override-2", "override-server", s, items} }) wg.Go(func() { events := []types.TransformerEvent{makeEvent("msg-combined-cached-2", versionCached)} - s, items := sendRawTransform(t, pyURL, events) + s, _, items := sendRawTransform(t, pyURL, events) results[3] = result{"cached-2", "cached-server", s, items} }) wg.Wait() @@ -435,28 +435,15 @@ func newMockAPIServer(t *testing.T, name string) (*httptest.Server, *atomic.Int6 return srv, calls } -// sendRawTransform sends events directly to pytransformer's /customTransform -// endpoint and returns the HTTP status code and parsed response items. -// Unlike the usertransformer.Client, this allows inspecting raw HTTP status. +// sendRawTransform sends events directly to pytransformer's /customTransform endpoint and +// returns the HTTP status code, the response headers and the parsed response items. Unlike the +// usertransformer.Client, this allows inspecting the raw HTTP status and the retry-contract +// headers (X-Rudder-Should-Retry / X-Rudder-Error-Reason); discard the headers with _ when a +// test does not assert on them. func sendRawTransform( t *testing.T, baseURL string, events []types.TransformerEvent, -) ( - int, - []types.TransformerResponse, -) { - t.Helper() - status, _, items := sendRawTransformWithHeaders(t, baseURL, events) - return status, items -} - -// sendRawTransformWithHeaders is sendRawTransform plus the response headers, for tests that -// assert the retry contract (X-Rudder-Should-Retry / X-Rudder-Error-Reason). -func sendRawTransformWithHeaders( - t *testing.T, - baseURL string, - events []types.TransformerEvent, ) ( int, http.Header, diff --git a/integration_test/pytransformer_contract/force_thread_pool_always_contract_test.go b/integration_test/pytransformer_contract/force_thread_pool_always_contract_test.go index 19e0724df..0be21c7a2 100644 --- a/integration_test/pytransformer_contract/force_thread_pool_always_contract_test.go +++ b/integration_test/pytransformer_contract/force_thread_pool_always_contract_test.go @@ -86,7 +86,7 @@ def transformEvent(event, metadata): { ev := makeEvent("single-evt", versionID) ev.Message["n"] = 1 - status, items := sendRawTransform(t, pyURL, []types.TransformerEvent{ev}) + status, _, items := sendRawTransform(t, pyURL, []types.TransformerEvent{ev}) require.Equal(t, http.StatusOK, status, "single-event /customTransform failed") require.Len(t, items, 1) require.Equalf(t, http.StatusOK, items[0].StatusCode, @@ -116,7 +116,7 @@ def transformEvent(event, metadata): batch[j] = ev inputs = append(inputs, inputEvent{messageID: messageID, n: n}) } - status, items := sendRawTransform(t, pyURL, batch) + status, _, items := sendRawTransform(t, pyURL, batch) require.Equalf(t, http.StatusOK, status, "request %d /customTransform failed", r) require.Lenf(t, items, eventsPerRequest, "request %d expected %d items", r, eventsPerRequest) diff --git a/integration_test/pytransformer_contract/http_timeout_conn_pool_contract_test.go b/integration_test/pytransformer_contract/http_timeout_conn_pool_contract_test.go index 598b62e4b..77c320c35 100644 --- a/integration_test/pytransformer_contract/http_timeout_conn_pool_contract_test.go +++ b/integration_test/pytransformer_contract/http_timeout_conn_pool_contract_test.go @@ -191,7 +191,7 @@ def transformEvent(event, metadata): // Our cap (1s) fires first. The transformation fails with a 400 // per-event error (non-retryable user code HTTP timeout). events := []types.TransformerEvent{makeEvent("msg-bigger-1", versionBiggerTimeout)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status, "/customTransform HTTP response must be 200 (per-event errors are in the payload)") require.Len(t, items, 1) @@ -215,7 +215,7 @@ def transformEvent(event, metadata): // The user passes timeout=0.1 s; the server replies after 2s. // The user's cap fires first (0.5s < our 1s cap < server's 2s delay). events := []types.TransformerEvent{makeEvent("msg-smaller-1", versionSmallerTimeout)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusBadRequest, items[0].StatusCode, @@ -272,13 +272,13 @@ def transformEvent(event, metadata): newConns.Store(0) ev1 := makeEvent("msg-pool-1", versionID) - status1, items1 := sendRawTransform(t, poolURL, []types.TransformerEvent{ev1}) + status1, _, items1 := sendRawTransform(t, poolURL, []types.TransformerEvent{ev1}) require.Equal(t, http.StatusOK, status1) require.Len(t, items1, 1) require.Equal(t, http.StatusOK, items1[0].StatusCode, "first request must succeed") ev2 := makeEvent("msg-pool-2", versionID) - status2, items2 := sendRawTransform(t, poolURL, []types.TransformerEvent{ev2}) + status2, _, items2 := sendRawTransform(t, poolURL, []types.TransformerEvent{ev2}) require.Equal(t, http.StatusOK, status2) require.Len(t, items2, 1) require.Equal(t, http.StatusOK, items2[0].StatusCode, "second request must succeed") @@ -352,13 +352,13 @@ def transformEvent(event, metadata): newConns.Store(0) ev1 := makeEvent("msg-shared-alpha", versionIDAlpha) - status1, items1 := sendRawTransform(t, sharedURL, []types.TransformerEvent{ev1}) + status1, _, items1 := sendRawTransform(t, sharedURL, []types.TransformerEvent{ev1}) require.Equal(t, http.StatusOK, status1) require.Len(t, items1, 1) require.Equal(t, http.StatusOK, items1[0].StatusCode, "alpha request must succeed") ev2 := makeEvent("msg-shared-beta", versionIDBeta) - status2, items2 := sendRawTransform(t, sharedURL, []types.TransformerEvent{ev2}) + status2, _, items2 := sendRawTransform(t, sharedURL, []types.TransformerEvent{ev2}) require.Equal(t, http.StatusOK, status2) require.Len(t, items2, 1) require.Equal(t, http.StatusOK, items2[0].StatusCode, "beta request must succeed") @@ -424,7 +424,7 @@ def transformEvent(event, metadata): events := []types.TransformerEvent{makeEvent("msg-slow-drip-1", versionID)} start := time.Now() - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) elapsed := time.Since(start) t.Logf("slow-drip request elapsed: %s", elapsed) diff --git a/integration_test/pytransformer_contract/io_bound_flag_cache_contract_test.go b/integration_test/pytransformer_contract/io_bound_flag_cache_contract_test.go index c45c6b4be..118504738 100644 --- a/integration_test/pytransformer_contract/io_bound_flag_cache_contract_test.go +++ b/integration_test/pytransformer_contract/io_bound_flag_cache_contract_test.go @@ -113,7 +113,7 @@ def transformEvent(event, metadata): // 1) Prime the only L1 slot with a CPU-bound transformation. fillerEvent := makeEvent("filler-1", fillerVersionID) - status, items := sendRawTransform(t, pyURL, []types.TransformerEvent{fillerEvent}) + status, _, items := sendRawTransform(t, pyURL, []types.TransformerEvent{fillerEvent}) require.Equal(t, http.StatusOK, status, "filler request must return 200") require.Len(t, items, 1, "filler request must produce one response item") require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -125,7 +125,7 @@ def transformEvent(event, metadata): for i := range eventsPerBatch { ioBatch1[i] = makeEvent(fmt.Sprintf("io1-%d", i), ioVersionID) } - status, items = sendRawTransform(t, pyURL, ioBatch1) + status, _, items = sendRawTransform(t, pyURL, ioBatch1) require.Equal(t, http.StatusOK, status, "first I/O batch must return 200") require.Len(t, items, eventsPerBatch, "first I/O batch must produce one item per event") for i, item := range items { @@ -148,7 +148,7 @@ def transformEvent(event, metadata): for i := range eventsPerBatch { ioBatch2[i] = makeEvent(fmt.Sprintf("io2-%d", i), ioVersionID) } - status, items = sendRawTransform(t, pyURL, ioBatch2) + status, _, items = sendRawTransform(t, pyURL, ioBatch2) require.Equal(t, http.StatusOK, status, "second I/O batch must return 200") require.Len(t, items, eventsPerBatch, "second I/O batch must produce one item per event") for i, item := range items { @@ -255,7 +255,7 @@ def transformEvent(event, metadata): ev.Message["do_http"] = true ioBatch[i] = ev } - status, items := sendRawTransform(t, pyURL, ioBatch) + status, _, items := sendRawTransform(t, pyURL, ioBatch) require.Equal(t, http.StatusOK, status, "I/O batch must return 200") require.Len(t, items, eventsPerBatch) for i, item := range items { @@ -274,7 +274,7 @@ def transformEvent(event, metadata): ev.Message["do_http"] = false cpuBatch[i] = ev } - status, items = sendRawTransform(t, pyURL, cpuBatch) + status, _, items = sendRawTransform(t, pyURL, cpuBatch) require.Equal(t, http.StatusOK, status, "CPU batch must return 200") require.Len(t, items, eventsPerBatch) for i, item := range items { diff --git a/integration_test/pytransformer_contract/managed_session_contract_test.go b/integration_test/pytransformer_contract/managed_session_contract_test.go index 1054c2b96..baeec1951 100644 --- a/integration_test/pytransformer_contract/managed_session_contract_test.go +++ b/integration_test/pytransformer_contract/managed_session_contract_test.go @@ -98,7 +98,7 @@ def transformEvent(event, metadata): ) events := []types.TransformerEvent{makeEvent("msg-module-retry-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -190,7 +190,7 @@ def transformEvent(event, metadata): ) events := []types.TransformerEvent{makeEvent("msg-call-budget-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -278,7 +278,7 @@ def transformEvent(event, metadata): ) events := []types.TransformerEvent{makeEvent("msg-verify-false-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -331,7 +331,7 @@ def transformEvent(event, metadata): ) events := []types.TransformerEvent{makeEvent("msg-req-verify-false-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -391,7 +391,7 @@ def transformEvent(event, metadata): hits.Store(0) events := []types.TransformerEvent{makeEvent("msg-subclass-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -455,7 +455,7 @@ def transformEvent(event, metadata): hitsA.Store(0) hitsB.Store(0) events := []types.TransformerEvent{makeEvent("msg-mount-prefix-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -514,7 +514,7 @@ def transformEvent(event, metadata): observed.reset() events := []types.TransformerEvent{makeEvent("msg-headers-auth-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -576,7 +576,7 @@ def transformEvent(event, metadata): newConns.Store(0) events := []types.TransformerEvent{makeEvent("msg-close-noop-1", versionID)} - status, items := sendRawTransform(t, pyURL, events) + status, _, items := sendRawTransform(t, pyURL, events) require.Equal(t, http.StatusOK, status) require.Len(t, items, 1) require.Equal(t, http.StatusOK, items[0].StatusCode, @@ -639,13 +639,13 @@ def transformEvent(event, metadata): newConns.Store(0) evA := makeEvent("msg-mgd-alpha", versionIDAlpha) - statusA, itemsA := sendRawTransform(t, pyURL, []types.TransformerEvent{evA}) + statusA, _, itemsA := sendRawTransform(t, pyURL, []types.TransformerEvent{evA}) require.Equal(t, http.StatusOK, statusA) require.Len(t, itemsA, 1) require.Equal(t, http.StatusOK, itemsA[0].StatusCode) evB := makeEvent("msg-mgd-beta", versionIDBeta) - statusB, itemsB := sendRawTransform(t, pyURL, []types.TransformerEvent{evB}) + statusB, _, itemsB := sendRawTransform(t, pyURL, []types.TransformerEvent{evB}) require.Equal(t, http.StatusOK, statusB) require.Len(t, itemsB, 1) require.Equal(t, http.StatusOK, itemsB[0].StatusCode) @@ -766,7 +766,7 @@ def transformEvent(event, metadata): for i := range evs { evs[i] = makeEvent(fmt.Sprintf("msg-%s-%d", variant.name, i), versionID) } - status, items := sendRawTransform(t, pyURL, evs) + status, _, items := sendRawTransform(t, pyURL, evs) require.Equal(t, http.StatusOK, status) require.Len(t, items, eventsPerRun) for i, it := range items { @@ -888,7 +888,7 @@ def transformEvent(event, metadata): for i := range evs { evs[i] = makeEvent(fmt.Sprintf("msg-pool-maxsize-%d", i), versionID) } - status, items := sendRawTransform(t, pyURL, evs) + status, _, items := sendRawTransform(t, pyURL, evs) require.Equal(t, http.StatusOK, status) require.Len(t, items, eventsPerRun) for i, it := range items { diff --git a/integration_test/pytransformer_contract/redirects_test.go b/integration_test/pytransformer_contract/redirects_test.go index 7922d85d8..fc849f000 100644 --- a/integration_test/pytransformer_contract/redirects_test.go +++ b/integration_test/pytransformer_contract/redirects_test.go @@ -47,7 +47,7 @@ func TestConfigBackendRedirectResponse(t *testing.T) { cb := newRedirectingConfigBackend(t, redirectStatus) pyURL := startRudderPytransformer(t, pool, cb.backend.URL) - status, headers, items := sendRawTransformWithHeaders(t, pyURL, makeEvents(versionID, 1)) + status, headers, items := sendRawTransform(t, pyURL, makeEvents(versionID, 1)) t.Logf("pytransformer returned HTTP %d, should-retry=%q reason=%q", status, headers.Get("X-Rudder-Should-Retry"), headers.Get("X-Rudder-Error-Reason")) diff --git a/integration_test/pytransformer_contract/requests_api_contract_test.go b/integration_test/pytransformer_contract/requests_api_contract_test.go index 85e59289c..a2cf5c106 100644 --- a/integration_test/pytransformer_contract/requests_api_contract_test.go +++ b/integration_test/pytransformer_contract/requests_api_contract_test.go @@ -222,13 +222,13 @@ def transformEvent(event, metadata): newConns.Store(0) ev1 := makeEvent("msg-reuse-1", rc.versionID) - status1, items1 := sendRawTransform(t, pyTransformerURL, []types.TransformerEvent{ev1}) + status1, _, items1 := sendRawTransform(t, pyTransformerURL, []types.TransformerEvent{ev1}) require.Equal(t, http.StatusOK, status1) require.Len(t, items1, 1) require.Equal(t, http.StatusOK, items1[0].StatusCode, "first request must succeed") ev2 := makeEvent("msg-reuse-2", rc.versionID) - status2, items2 := sendRawTransform(t, pyTransformerURL, []types.TransformerEvent{ev2}) + status2, _, items2 := sendRawTransform(t, pyTransformerURL, []types.TransformerEvent{ev2}) require.Equal(t, http.StatusOK, status2) require.Len(t, items2, 1) require.Equal(t, http.StatusOK, items2[0].StatusCode, "second request must succeed") diff --git a/integration_test/pytransformer_contract/threaded_execution_contract_test.go b/integration_test/pytransformer_contract/threaded_execution_contract_test.go index c38fdbfed..eadf2b3e5 100644 --- a/integration_test/pytransformer_contract/threaded_execution_contract_test.go +++ b/integration_test/pytransformer_contract/threaded_execution_contract_test.go @@ -146,7 +146,7 @@ def transformEvent(event, metadata): warmupID := fmt.Sprintf("warmup-%d", i) warmupEvent := makeEvent(warmupID, versionID) warmupEvent.Message["marker"] = warmupID - _, items := sendRawTransform(t, pyURL, []types.TransformerEvent{warmupEvent}) + _, _, items := sendRawTransform(t, pyURL, []types.TransformerEvent{warmupEvent}) require.Len(t, items, 1, "warmup %d: expected one response item", i) require.Equal(t, http.StatusOK, items[0].StatusCode, "warmup %d: expected HTTP 200 (error=%s)", i, items[0].Error) @@ -161,7 +161,7 @@ def transformEvent(event, metadata): ) for i := range parallelRequests { wg.Go(func() { - status, items := sendRawTransform(t, pyURL, requests[i]) + status, _, items := sendRawTransform(t, pyURL, requests[i]) results[i] = result{idx: i, status: status, items: items} }) } From 12900c0b244b2115fc936010291991821a882cc1 Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Mon, 3 Aug 2026 10:43:38 +0200 Subject: [PATCH 6/7] chore: pyt config backend auth contract tests --- .../config_backend_auth_test.go | 715 ++++++++++++++++++ 1 file changed, 715 insertions(+) create mode 100644 integration_test/pytransformer_contract/config_backend_auth_test.go diff --git a/integration_test/pytransformer_contract/config_backend_auth_test.go b/integration_test/pytransformer_contract/config_backend_auth_test.go new file mode 100644 index 000000000..35a32ee36 --- /dev/null +++ b/integration_test/pytransformer_contract/config_backend_auth_test.go @@ -0,0 +1,715 @@ +package pytransformer_contract + +import ( + "crypto/subtle" + "encoding/base64" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/ory/dockertest/v3" + "github.com/stretchr/testify/require" + + "github.com/rudderlabs/rudder-go-kit/jsonrs" +) + +// This file pins what CONFIG_BACKEND_HOSTED_SECRET does to the wire, both variants, against a +// mock config backend ported from ../rudder-config-backend rather than invented here (see +// configBackendAuthMock below for the file-by-file provenance). +// +// The question it answers first is the boring one: a deployment that never sets the variable +// must be indistinguishable from the build that did not know about it. "We return {} when the +// secret is unset" is an implementation detail; what a deployment cares about is that the +// request reaching the config backend is unchanged, which is what +// TestConfigBackendAuthWithoutHostedSecret asserts directly — no Authorization header at all, +// same path, same query, events transformed. +// +// The rest is the other variant: with the secret set, the header is exactly +// Basic base64(":") — the empty password is load-bearing, rudder-config-backend rejects +// any decoded token that does not hold exactly one colon — and a wrong secret surfaces as a +// retryable 503 rather than a per-event failure rudder-server would abort. + +const ( + // cbAuthSecret is the secret both pytransformer and the mock config backend are given in + // the happy path. Restricted to [A-Za-z0-9_-] on purpose: a ":" breaks the basic-auth token + // and a "," is split apart by the config backend's comma-separated secret list. + cbAuthSecret = "py-contract-hosted-secret" + + // cbAuthOtherSecret is a second, valid secret configured on the config backend alongside + // cbAuthSecret. rudder-config-backend accepts a comma-separated list so a secret can be + // rotated without a flag day; carrying two here keeps that path exercised. + cbAuthOtherSecret = "py-contract-hosted-secret-next" + + // cbAuthTransformationRoute / cbAuthLibraryRoute are the two routes pytransformer fetches + // from. Both are authenticated on the internal gateway, and the library one is easy to + // forget — it is a second call site for the header. + cbAuthTransformationRoute = "/transformation/getByVersionId" + cbAuthLibraryRoute = "/transformationLibrary/getByVersionId" +) + +// cbAuthCode sets event["foo"] = "bar", the marker every happy-path assertion looks for. +const cbAuthCode = ` +def transformEvent(event, metadata): + event['foo'] = 'bar' + return event +` + +// cbAuthCodeUsingLibrary imports cbAuthLibraryCode, so a request using it only comes back +// transformed if the *library* fetch authenticated too — a second call site for the header. +const ( + cbAuthLibraryImportName = "cbauthlib" + cbAuthLibraryCode = ` +def marker(): + return 'bar-from-library' +` + cbAuthCodeUsingLibrary = ` +import cbauthlib + +def transformEvent(event, metadata): + event['foo'] = cbauthlib.marker() + return event +` +) + +// TestConfigBackendAuthWithoutHostedSecret is the regression guard for the change being +// additive: with CONFIG_BACKEND_HOSTED_SECRET absent from the environment, pytransformer must +// talk to the config backend exactly as the build before it did. +// +// The assertion that carries the claim is not "it still works" — it is that every request the +// config backend received arrived with no Authorization header whatsoever. A build that sent +// `Authorization: Basic Og==` (base64 of ":") would still transform events fine against a +// permissive backend and quietly break against a strict one; only inspecting the received +// header separates the two. +func TestConfigBackendAuthWithoutHostedSecret(t *testing.T) { + pool, err := dockertest.NewPool("") + require.NoError(t, err) + + cb := newConfigBackendAuthMock(t, cbAuthSecret, cbAuthOtherSecret) + + // No CONFIG_BACKEND_HOSTED_SECRET passed at all — not empty, absent. This is the + // pre-change deployment, and every self-hosted one. + pyURL := startRudderPytransformer(t, pool, cb.server.URL) + + t.Run("public unauthenticated routes are reached with no Authorization header", func(t *testing.T) { + const versionID = "cbauth-nosecret-public-v1" + cb.setMode(cbModePublic) + + status, headers, items := sendRawTransform(t, pyURL, makeEvents(versionID, 2)) + + require.Equal(t, http.StatusOK, status) + require.Empty(t, headers.Get("X-Rudder-Should-Retry")) + require.Len(t, items, 2) + for _, item := range items { + require.Equal(t, http.StatusOK, item.StatusCode, "error: %s", item.Error) + require.Equal(t, "bar", item.Output["foo"]) + } + + // The load-bearing part. Not "the header was wrong" — the header was not there. + seen := cb.requestsFor(cbAuthTransformationRoute, versionID) + require.NotEmpty(t, seen, "config backend was never asked for the transformation code") + for _, r := range seen { + require.Empty(t, r.authorization, + "an unset CONFIG_BACKEND_HOSTED_SECRET must send no Authorization header at all") + require.Equal(t, versionID, r.versionID) + } + }) + + t.Run("hosted-secret routes reject the anonymous fetch as retryable", func(t *testing.T) { + // Pointing CONFIG_BACKEND_URL at the internal gateway and forgetting the secret. The + // interesting question is not that it fails but *how*: a per-event 401 is terminal, and + // rudder-server aborts every per-event status that is not 200/298, so the events would + // be destroyed by a configuration mistake. + const versionID = "cbauth-nosecret-authenticated-v1" + cb.setMode(cbModeHostedSecret) + + status, headers, items := sendRawTransform(t, pyURL, makeEvents(versionID, 2)) + + require.Equal(t, http.StatusServiceUnavailable, status) + require.Equal(t, "true", headers.Get("X-Rudder-Should-Retry")) + require.Equal(t, "config_backend_auth_failed", headers.Get("X-Rudder-Error-Reason")) + require.Len(t, items, 2) + for _, item := range items { + require.Equal(t, http.StatusServiceUnavailable, item.StatusCode) + require.Contains(t, item.Error, "401") + } + + seen := cb.requestsFor(cbAuthTransformationRoute, versionID) + require.NotEmpty(t, seen) + for _, r := range seen { + require.Empty(t, r.authorization) + require.Equal(t, http.StatusUnauthorized, r.status) + } + }) + + t.Run("a blocked public route surfaces as forbidden, not as auth failed", func(t *testing.T) { + // rudder-config-backend's blockHostedPublicAccess answers 403 for a paid workspace once + // BLOCK_PUBLIC_TRANSFORMATION_ROUTES is on. Same retryable treatment, different reason, + // because "the secret is wrong" and "this pod never moved to the internal gateway" page + // different people. + const versionID = "cbauth-nosecret-blocked-v1" + cb.setMode(cbModePublicBlocked) + + status, headers, items := sendRawTransform(t, pyURL, makeEvents(versionID, 1)) + + require.Equal(t, http.StatusServiceUnavailable, status) + require.Equal(t, "true", headers.Get("X-Rudder-Should-Retry")) + require.Equal(t, "config_backend_forbidden", headers.Get("X-Rudder-Error-Reason")) + require.Len(t, items, 1) + require.Equal(t, http.StatusServiceUnavailable, items[0].StatusCode) + require.Contains(t, items[0].Error, "403") + }) +} + +// TestConfigBackendAuthWithHostedSecret is the other variant: the secret is set, so every fetch +// must carry Basic base64(":") and authenticate against a config backend that enforces +// the real check. +func TestConfigBackendAuthWithHostedSecret(t *testing.T) { + pool, err := dockertest.NewPool("") + require.NoError(t, err) + + cb := newConfigBackendAuthMock(t, cbAuthSecret, cbAuthOtherSecret) + pyURL := startRudderPytransformer(t, pool, cb.server.URL, + "CONFIG_BACKEND_HOSTED_SECRET="+cbAuthSecret) + + // What the config backend must receive, spelled out rather than recomputed from the same + // helper the assertion is checking: the secret is the username and the password is empty. + wantHeader := "Basic " + base64.StdEncoding.EncodeToString([]byte(cbAuthSecret+":")) + + t.Run("the transformation fetch authenticates", func(t *testing.T) { + const versionID = "cbauth-secret-transformation-v1" + cb.setMode(cbModeHostedSecret) + + status, headers, items := sendRawTransform(t, pyURL, makeEvents(versionID, 2)) + + require.Equal(t, http.StatusOK, status, "items: %+v", items) + require.Empty(t, headers.Get("X-Rudder-Should-Retry")) + require.Len(t, items, 2) + for _, item := range items { + require.Equal(t, http.StatusOK, item.StatusCode, "error: %s", item.Error) + require.Equal(t, "bar", item.Output["foo"]) + } + + seen := cb.requestsFor(cbAuthTransformationRoute, versionID) + require.NotEmpty(t, seen) + for _, r := range seen { + require.Equal(t, wantHeader, r.authorization) + require.Equal(t, http.StatusOK, r.status) + } + }) + + t.Run("the library fetch carries the same header", func(t *testing.T) { + // A second call site that is easy to miss. The transformation imports the library, so + // the event only comes back transformed if the library fetch authenticated too — the + // header assertion below is corroboration, not the only evidence. + const ( + versionID = "cbauth-secret-library-v1" + libVersionID = "cbauth-secret-library-lib-v1" + ) + cb.setMode(cbModeHostedSecret) + cb.addTransformation(versionID, cbAuthCodeUsingLibrary) + cb.addLibrary(libVersionID, cbAuthLibraryImportName, cbAuthLibraryCode) + + status, _, items := sendRawTransform(t, pyURL, + makeEvents(versionID, 1, libVersionID)) + + require.Equal(t, http.StatusOK, status, "items: %+v", items) + require.Len(t, items, 1) + require.Equal(t, http.StatusOK, items[0].StatusCode, "error: %s", items[0].Error) + require.Equal(t, "bar-from-library", items[0].Output["foo"]) + + seen := cb.requestsFor(cbAuthLibraryRoute, libVersionID) + require.NotEmpty(t, seen, "config backend was never asked for the library") + for _, r := range seen { + require.Equal(t, wantHeader, r.authorization) + require.Equal(t, http.StatusOK, r.status) + } + }) + + t.Run("a public config backend ignores the header and still serves", func(t *testing.T) { + // Setting the secret while still pointed at the public routes is a harmless + // intermediate state during a repoint, and must stay harmless: the header is sent, the + // unauthenticated route does not look at it. + const versionID = "cbauth-secret-public-v1" + cb.setMode(cbModePublic) + + status, _, items := sendRawTransform(t, pyURL, makeEvents(versionID, 1)) + + require.Equal(t, http.StatusOK, status, "items: %+v", items) + require.Len(t, items, 1) + require.Equal(t, http.StatusOK, items[0].StatusCode, "error: %s", items[0].Error) + require.Equal(t, "bar", items[0].Output["foo"]) + + seen := cb.requestsFor(cbAuthTransformationRoute, versionID) + require.NotEmpty(t, seen) + require.Equal(t, wantHeader, seen[0].authorization) + }) +} + +// TestConfigBackendAuthWithWrongHostedSecret is the unhappy path that matters operationally: a +// stale secret after a rotation. It must be retryable, so the events survive until someone +// updates the secret, exactly like the no-secret-against-a-strict-backend case. +func TestConfigBackendAuthWithWrongHostedSecret(t *testing.T) { + pool, err := dockertest.NewPool("") + require.NoError(t, err) + + cb := newConfigBackendAuthMock(t, cbAuthSecret, cbAuthOtherSecret) + cb.setMode(cbModeHostedSecret) + pyURL := startRudderPytransformer(t, pool, cb.server.URL, + "CONFIG_BACKEND_HOSTED_SECRET=py-contract-hosted-secret-rotated-away") + + const versionID = "cbauth-wrongsecret-v1" + status, headers, items := sendRawTransform(t, pyURL, makeEvents(versionID, 3)) + + require.Equal(t, http.StatusServiceUnavailable, status) + require.Equal(t, "true", headers.Get("X-Rudder-Should-Retry")) + require.Equal(t, "config_backend_auth_failed", headers.Get("X-Rudder-Error-Reason")) + require.Len(t, items, 3) + for _, item := range items { + require.Equal(t, http.StatusServiceUnavailable, item.StatusCode) + } + + // The header was sent and was well-formed — it simply is not one of the configured + // secrets. Without this the test could pass on a request that never carried a header. + seen := cb.requestsFor(cbAuthTransformationRoute, versionID) + require.NotEmpty(t, seen) + for _, r := range seen { + require.Equal(t, + "Basic "+base64.StdEncoding.EncodeToString([]byte("py-contract-hosted-secret-rotated-away:")), + r.authorization) + require.Equal(t, http.StatusUnauthorized, r.status) + } +} + +// TestConfigBackendHostedSecretTrailingNewlineIsTrimmed covers the shape the secret actually +// arrives in on Kubernetes: mounted from a file, which commonly ends in a newline. Untrimmed it +// would 401 every fetch, and the failure would look like a wrong secret rather than a stray byte. +func TestConfigBackendHostedSecretTrailingNewlineIsTrimmed(t *testing.T) { + pool, err := dockertest.NewPool("") + require.NoError(t, err) + + cb := newConfigBackendAuthMock(t, cbAuthSecret) + cb.setMode(cbModeHostedSecret) + pyURL := startRudderPytransformer(t, pool, cb.server.URL, + "CONFIG_BACKEND_HOSTED_SECRET="+cbAuthSecret+"\n") + + const versionID = "cbauth-trailing-newline-v1" + status, _, items := sendRawTransform(t, pyURL, makeEvents(versionID, 1)) + + require.Equal(t, http.StatusOK, status, "items: %+v", items) + require.Len(t, items, 1) + require.Equal(t, http.StatusOK, items[0].StatusCode, "error: %s", items[0].Error) + require.Equal(t, "bar", items[0].Output["foo"]) + + seen := cb.requestsFor(cbAuthTransformationRoute, versionID) + require.NotEmpty(t, seen) + require.Equal(t, + "Basic "+base64.StdEncoding.EncodeToString([]byte(cbAuthSecret+":")), + seen[0].authorization, + "the newline must be stripped before the token is built, not encoded into it") +} + +// TestConfigBackendAuthMockMatchesConfigBackend pins the port itself. The container tests above +// are only worth their runtime if the thing they authenticate against rejects and accepts the +// same inputs rudder-config-backend does, and the rules are not obvious — the single-colon check +// and the trim-and-drop-blanks list parsing are both easy to approximate wrongly. +// +// Cases and expected outcomes are taken from rudder-config-backend's own suites: +// src/__tests__/config.test.ts and src/modules/rudder-basic-auth/__tests__. +func TestConfigBackendAuthMockMatchesConfigBackend(t *testing.T) { + basic := func(token string) string { + return "Basic " + base64.StdEncoding.EncodeToString([]byte(token)) + } + + t.Run("hostedSecretCredentials", func(t *testing.T) { + for _, tc := range []struct { + name string + raw string + expected []string + }{ + {"single secret", "s3cr3t", []string{"s3cr3t"}}, + {"comma separated list", "rotating-old,rotating-new", []string{"rotating-old", "rotating-new"}}, + {"entries are trimmed", " padded , also-padded ", []string{"padded", "also-padded"}}, + {"blank entries are dropped", "kept,, ,also-kept", []string{"kept", "also-kept"}}, + {"unset keeps the development default", "", []string{"password"}}, + } { + t.Run(tc.name, func(t *testing.T) { + creds := hostedSecretCredentials(tc.raw) + usernames := make([]string, len(creds)) + for i, c := range creds { + usernames[i] = c.username + require.Empty(t, c.password, + "every hosted secret is a username with an empty password") + } + require.Equal(t, tc.expected, usernames) + }) + } + }) + + t.Run("rudderKoaBasicAuth", func(t *testing.T) { + creds := hostedSecretCredentials("s3cr3t,rotating-new") + for _, tc := range []struct { + name string + header string + wantReason string + }{ + {"configured secret with an empty password", basic("s3cr3t:"), ""}, + {"any secret in the list authenticates", basic("rotating-new:"), ""}, + {"missing header", "", cbAuthMissingHeader}, + {"bearer instead of basic", "Bearer s3cr3t", cbAuthMissingHeader}, + // The single-colon rule. This is why pytransformer must send ":" and not + // a bare "": no colon at all is rejected before any comparison happens. + {"no colon in the decoded token", basic("s3cr3t"), cbAuthInvalidHeader}, + {"more than one colon", basic("s3cr3t:extra:colon"), cbAuthInvalidHeader}, + {"empty token", basic(""), cbAuthInvalidHeader}, + {"colon only", basic(":"), cbAuthInvalidCredentials}, + {"unknown secret", basic("not-the-secret:"), cbAuthInvalidCredentials}, + // Same length as "s3cr3t:", so it gets past the length filter and must still be + // rejected by the constant-time comparison. + {"same-length wrong secret", basic("s3cr3u:"), cbAuthInvalidCredentials}, + {"right secret, non-empty password", basic("s3cr3t:pw"), cbAuthInvalidCredentials}, + {"untrimmed secret in the token", basic(" s3cr3t :"), cbAuthInvalidCredentials}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.wantReason, rudderKoaBasicAuth(creds, tc.header)) + }) + } + }) +} + +// --------------------------------------------------------------------------- +// Mock config backend +// --------------------------------------------------------------------------- + +// cbAuthMode selects which of rudder-config-backend's route families the mock imitates for the +// two fetch paths. The same handler serves all three because in production it is the same two +// URLs — only the router they were registered on differs, and that is a deploy-time choice a +// pytransformer pod cannot see. +type cbAuthMode int32 + +const ( + // cbModePublic is the deprecated public router in + // ../rudder-config-backend/src/modules/transformations/routes.ts (unAuthenticatedRouter): + // no authentication of any kind. This is what api.rudderlabs.com serves today and what + // every self-hosted config backend serves. + cbModePublic cbAuthMode = iota + + // cbModeHostedSecret is the internal gateway in + // ../rudder-config-backend/src/modules/transformations/internalRoutes.ts, registered on + // getDataplaneGatewayRouter, which attaches DataPlaneController.verifyHostedDataPlaneSecret + // to every route. Same paths, so a caller only swaps the base URL — and must start sending + // the hosted secret. + cbModeHostedSecret + + // cbModePublicBlocked is the public router with BLOCK_PUBLIC_TRANSFORMATION_ROUTES on and + // the versionId's workspace on a paid plan: DataPlaneController.blockHostedPublicAccess + // throws ForbiddenError(Message.Unauthorized). The 403 is how a pod that never moved to the + // internal gateway finds out. + cbModePublicBlocked +) + +// cbCredential is one username/password pair, mirroring the Credential type in +// ../rudder-config-backend/src/modules/rudder-basic-auth/index.ts. +type cbCredential struct { + username string + password string +} + +// cbAuthRequest is one request the mock received, kept so a test can assert on what was +// actually on the wire rather than on what the code under test says it sends. +type cbAuthRequest struct { + path string + versionID string + authorization string + status int +} + +// configBackendAuthMock is a mock rudder-config-backend for the two routes pytransformer +// fetches from, ported from ../rudder-config-backend so the authentication it enforces is the +// authentication pytransformer meets in production. Ported pieces, in the order a request meets +// them: +// +// - src/modules/transformations/internalRoutes.ts — the internal-gateway mirror of +// /transformation/getByVersionId and /transformationLibrary/getByVersionId, registered on +// getDataplaneGatewayRouter. +// - src/modules/internal-gateway/router.ts — getDataplaneGatewayRouter attaches +// DataPlaneController.verifyHostedDataPlaneSecret to every route on it. +// - src/controllers/dataPlane.controller.ts — verifyHostedDataPlaneSecret swallows the +// specific failure and throws UnauthenticatedError(Message.IncorrectHostedServiceSecret), +// so every rejection reads the same regardless of which rule was broken. +// - src/modules/rudder-basic-auth/index.ts — RudderKoaBasicAuth, the actual check (see +// rudderKoaBasicAuth). +// - src/config.ts — hostedSecretConfig (see hostedSecretCredentials). +// - src/serverUtils/middlewares.ts — a thrown CustomError becomes {"message": ...} with the +// error's own status code, and no WWW-Authenticate header. +// - src/controllers/transformation.controller.ts and transformationLibrary.controller.ts — +// the success bodies and the 400 for a versionId that resolves to nothing. +// +// Deliberately not ported: the workspace-token alternative on /data-plane/v1 (pytransformer +// never holds a workspace token) and the leniency of Node's base64 decoder (see +// rudderKoaBasicAuth). +type configBackendAuthMock struct { + server *httptest.Server + + // hostedSecrets is the parsed HOSTED_SERVICE_SECRETS of the config backend being imitated. + hostedSecrets []cbCredential + + mode atomic.Int32 + + mu sync.Mutex + transformations map[string]string // versionId -> code + libraries map[string]cbLibrary // versionId -> library + received []cbAuthRequest +} + +type cbLibrary struct { + importName string + code string +} + +// newConfigBackendAuthMock starts a mock config backend whose HOSTED_SERVICE_SECRETS is the +// given list. It starts in cbModePublic; a test picks the mode it needs with setMode. +// +// Every versionId resolves — an unregistered one is served cbAuthCode — so a test only registers +// a transformation when the code itself matters (the library case). Subtests still need distinct +// versionIds, but for a different reason: pytransformer caches fetched code in its L2 cache and +// would not re-fetch, so a reused id would be answered without the config backend being asked. +func newConfigBackendAuthMock(t *testing.T, hostedServiceSecrets ...string) *configBackendAuthMock { + t.Helper() + + cb := &configBackendAuthMock{ + hostedSecrets: hostedSecretCredentials(strings.Join(hostedServiceSecrets, ",")), + transformations: map[string]string{}, + libraries: map[string]cbLibrary{}, + } + + cb.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + versionID := r.URL.Query().Get("versionId") + authorization := r.Header.Get("Authorization") + + status, body := cb.handle(r.URL.Path, versionID, authorization) + + cb.mu.Lock() + cb.received = append(cb.received, cbAuthRequest{ + path: r.URL.Path, + versionID: versionID, + authorization: authorization, + status: status, + }) + cb.mu.Unlock() + + t.Logf("ConfigBackend: %s versionId=%q authorization=%q -> %d", + r.URL.Path, versionID, authorization, status) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + if err := jsonrs.NewEncoder(w).Encode(body); err != nil { + t.Errorf("ConfigBackend: failed to encode response: %v", err) + } + })) + t.Cleanup(cb.server.Close) + + return cb +} + +// handle is the request pipeline: route match, then the mode's authentication, then the +// controller. The order matters and is the config backend's — authentication runs as middleware, +// so a request with a bad secret gets 401 even when the versionId is also unknown. +func (cb *configBackendAuthMock) handle(path, versionID, authorization string) (int, any) { + if path != cbAuthTransformationRoute && path != cbAuthLibraryRoute { + return http.StatusNotFound, map[string]any{"message": "Not Found"} + } + + switch cbAuthMode(cb.mode.Load()) { + case cbModePublic: + // unAuthenticatedRouter: the Authorization header, present or not, is never read. + case cbModeHostedSecret: + if rudderKoaBasicAuth(cb.hostedSecrets, authorization) != "" { + // verifyHostedDataPlaneSecret discards the specific reason and throws + // UnauthenticatedError(Message.IncorrectHostedServiceSecret) -> 401. + return http.StatusUnauthorized, map[string]any{ + "message": "Incorrect hosted workspace secret", + } + } + case cbModePublicBlocked: + // blockHostedPublicAccess -> ForbiddenError(Message.Unauthorized) -> 403. The real one + // blocks only if the versionId's workspace resolves and is on a paid plan; here every + // versionId resolves and the mode is the paid-workspace posture, so it always blocks. + return http.StatusForbidden, map[string]any{"message": "Unauthorised"} + } + + if versionID == "" { + // TransformationController.getByVersionId: BadRequestError('versionId is required'). + return http.StatusBadRequest, map[string]any{"message": "versionId is required"} + } + + cb.mu.Lock() + defer cb.mu.Unlock() + + if path == cbAuthLibraryRoute { + lib, ok := cb.libraries[versionID] + if !ok { + return http.StatusBadRequest, map[string]any{ + "message": fmt.Sprintf( + "Transformation library not found for given version id: %s", versionID), + } + } + // TransformationLibraryController.getByVersionId: the library plus importName from + // getHandleName(). pytransformer reads importName and code. + return http.StatusOK, map[string]any{ + "versionId": versionID, + "name": lib.importName, + "handleName": lib.importName, + "importName": lib.importName, + "code": lib.code, + "language": "pythonfaas", + } + } + + code, ok := cb.transformations[versionID] + if !ok { + code = cbAuthCode + } + // TransformationController.getByVersionId: the revision DTO with secrets attached. + // pytransformer reads code; rudder-transformer also reads language and codeVersion. + return http.StatusOK, map[string]any{ + "versionId": versionID, + "name": "Config backend auth contract test", + "description": "", + "code": code, + "language": "pythonfaas", + "codeVersion": "1", + "secretsVersion": nil, + "imports": []any{}, + "secrets": map[string]any{}, + } +} + +// setMode switches which route family the mock imitates. Safe to call between subtests sharing +// one container; each subtest must use its own versionId, since pytransformer caches fetched +// code in its L2 cache and would not re-fetch. +func (cb *configBackendAuthMock) setMode(mode cbAuthMode) { + cb.mode.Store(int32(mode)) +} + +func (cb *configBackendAuthMock) addTransformation(versionID, code string) { + cb.mu.Lock() + defer cb.mu.Unlock() + cb.transformations[versionID] = code +} + +func (cb *configBackendAuthMock) addLibrary(versionID, importName, code string) { + cb.mu.Lock() + defer cb.mu.Unlock() + cb.libraries[versionID] = cbLibrary{importName: importName, code: code} +} + +// requestsFor returns every request the mock received for a path and versionId. Scoped by +// versionId because subtests share a mock, and an assertion that swept in a sibling subtest's +// requests would be reporting on the wrong configuration. +func (cb *configBackendAuthMock) requestsFor(path, versionID string) []cbAuthRequest { + cb.mu.Lock() + defer cb.mu.Unlock() + + var out []cbAuthRequest + for _, r := range cb.received { + if r.path == path && r.versionID == versionID { + out = append(out, r) + } + } + return out +} + +// hostedSecretCredentials ports parseHostedServiceSecrets and hostedSecretConfig from +// ../rudder-config-backend/src/config.ts: HOSTED_SERVICE_SECRETS is a comma-separated list, each +// entry trimmed and blanks dropped, and every password is the empty string. That last detail is +// the whole reason pytransformer sends base64(":") rather than base64(""). +// +// The unset case keeps the config backend's development default of a single "password" secret. +// Its startup-time throw for a set-but-all-blank value is not ported: it is a config backend +// startup concern with nothing for pytransformer to observe. +func hostedSecretCredentials(hostedServiceSecrets string) []cbCredential { + var usernames []string + for _, secret := range strings.Split(hostedServiceSecrets, ",") { + if trimmed := strings.TrimSpace(secret); trimmed != "" { + usernames = append(usernames, trimmed) + } + } + if len(usernames) == 0 { + usernames = []string{"password"} + } + + credentials := make([]cbCredential, len(usernames)) + for i, username := range usernames { + credentials[i] = cbCredential{username: username, password: ""} + } + return credentials +} + +// The three rejection messages RudderKoaBasicAuth attaches to its UnauthenticatedError, verbatim +// from ../rudder-config-backend/src/modules/rudder-basic-auth/index.ts. They are quotations, not +// Go error strings, hence the capitalisation. +const ( + cbAuthMissingHeader = "Authorization header is missing" + cbAuthInvalidHeader = "Invalid Authorization header" + cbAuthInvalidCredentials = "Invalid credentials" +) + +// rudderKoaBasicAuth ports RudderKoaBasicAuth from +// ../rudder-config-backend/src/modules/rudder-basic-auth/index.ts. It returns "" when the header +// authenticates, otherwise the message that rule would have thrown — a string rather than an +// error because it is never wrapped or compared, and because verifyHostedDataPlaneSecret +// collapses all three into one 401 body on the wire. Keeping them distinct here is what lets a +// test say *which* rule rejected a header. +// +// The rule worth naming is the colon count: the decoded token must contain exactly one ":". +// A bare "" is rejected before any secret comparison happens, which is why the empty +// password in ":" is not cosmetic. +// +// One deliberate divergence: Node's Buffer.from(x, 'base64') silently ignores characters outside +// the base64 alphabet, Go's decoder errors. Malformed base64 therefore takes the same branch here +// that garbage bytes would take there — rejected either way, and no caller sends malformed base64. +func rudderKoaBasicAuth(credentials []cbCredential, header string) string { + if header == "" { + return cbAuthMissingHeader + } + + // JS: const [type, base64token] = header.split(' ') — extra segments are dropped, and a + // header with no space leaves base64token undefined, which throws downstream. Either way + // the request is rejected. + parts := strings.Split(header, " ") + if parts[0] != "Basic" || len(parts) < 2 { + return cbAuthMissingHeader + } + + raw, err := base64.StdEncoding.DecodeString(parts[1]) + if err != nil { + return cbAuthInvalidHeader + } + + token := string(raw) + if !strings.Contains(token, ":") || strings.Index(token, ":") != strings.LastIndex(token, ":") { + return cbAuthInvalidHeader + } + + // Comparing every candidate of matching length without short-circuiting, as the original + // does: length is compared first because it leaks nothing, and the survivors go through a + // constant-time comparison whose results are summed rather than any-ed. + claim := []byte(token) + lengthMatches, credentialMatches := 0, 0 + for _, c := range credentials { + want := []byte(c.username + ":" + c.password) + if len(want) != len(claim) { + continue + } + lengthMatches++ + credentialMatches += subtle.ConstantTimeCompare(want, claim) + } + if lengthMatches == 0 || credentialMatches < 1 { + return cbAuthInvalidCredentials + } + return "" +} From 27fba11d548462cfc87a602fca7faa7c52be2c8c Mon Sep 17 00:00:00 2001 From: Francesco Casula Date: Mon, 3 Aug 2026 17:27:31 +0200 Subject: [PATCH 7/7] fix: linting --- .../pytransformer_contract/config_backend_auth_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integration_test/pytransformer_contract/config_backend_auth_test.go b/integration_test/pytransformer_contract/config_backend_auth_test.go index 35a32ee36..0b8e8c75f 100644 --- a/integration_test/pytransformer_contract/config_backend_auth_test.go +++ b/integration_test/pytransformer_contract/config_backend_auth_test.go @@ -633,7 +633,7 @@ func (cb *configBackendAuthMock) requestsFor(path, versionID string) []cbAuthReq // startup concern with nothing for pytransformer to observe. func hostedSecretCredentials(hostedServiceSecrets string) []cbCredential { var usernames []string - for _, secret := range strings.Split(hostedServiceSecrets, ",") { + for secret := range strings.SplitSeq(hostedServiceSecrets, ",") { if trimmed := strings.TrimSpace(secret); trimmed != "" { usernames = append(usernames, trimmed) }