package upload import ( "archive/tar" "compress/gzip" "context" "encoding/json" "errors" "io" "net/http" "net/http/httptest" "os" "path/filepath" "regexp" "strings" "sync/atomic" "testing" "time" sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle" ) func TestNewClientValidatesOptions(t *testing.T) { tests := []struct { name string opts ClientOptions }{ {name: "missing endpoint", opts: ClientOptions{Token: "secret"}}, {name: "missing token", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080"}}, {name: "bad scheme", opts: ClientOptions{Endpoint: "ftp://127.0.0.1:8080", Token: "secret"}}, {name: "missing host", opts: ClientOptions{Endpoint: "http:///upload", Token: "secret"}}, {name: "query", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080?x=1", Token: "secret"}}, {name: "userinfo", opts: ClientOptions{Endpoint: "http://user@127.0.0.1:8080", Token: "secret"}}, {name: "negative attempts", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080", Token: "secret", Retry: RetryOptions{MaxAttempts: -1}}}, {name: "negative delay", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080", Token: "secret", Retry: RetryOptions{BaseDelay: -1}}}, {name: "max below base", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080", Token: "secret", Retry: RetryOptions{BaseDelay: time.Second, MaxDelay: time.Millisecond}}}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { if _, err := NewClient(tt.opts); err == nil { t.Fatal("NewClient() error = nil, want error") } }) } client, err := NewClient(ClientOptions{Endpoint: "http://127.0.0.1:8080/base/", Token: "secret"}) if err != nil { t.Fatalf("NewClient() error = %v", err) } if got, want := client.uploadURL(), "http://127.0.0.1:8080/base/upload"; got != want { t.Fatalf("upload URL = %q, want %q", got, want) } if client.httpClient == nil || client.httpClient.Timeout == 0 { t.Fatalf("default HTTP client = %#v, want timeout", client.httpClient) } } func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{ {path: "report.md", data: "# Report\n"}, {path: "nested/summary.txt", data: "Summary\n"}, }) if err := os.WriteFile(filepath.Join(root, "unlisted.txt"), []byte("nope"), 0o600); err != nil { t.Fatalf("write unlisted file: %v", err) } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if got, want := r.URL.Path, "/upload"; got != want { t.Fatalf("path = %q, want %q", got, want) } if got, want := r.Header.Get("Authorization"), "Bearer secret-token"; got != want { t.Fatalf("authorization = %q, want %q", got, want) } if got, want := r.Header.Get("Content-Type"), uploadContentTypeGzip; got != want { t.Fatalf("content type = %q, want %q", got, want) } if got, want := r.Header.Get(idempotencyKeyHeader), "producer.retry:one"; got != want { t.Fatalf("idempotency key = %q, want %q", got, want) } entries := readArchiveEntries(t, r.Body) if got, want := strings.Join(entryNames(entries), ","), "manifest.json,report.md,nested/summary.txt"; got != want { t.Fatalf("archive entries = %q, want %q", got, want) } if _, ok := entries["unlisted.txt"]; ok { t.Fatal("archive included unlisted file") } writeAccepted(t, w, "reports.20260604T120000Z.abcdef12") })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret-token", HTTPClient: server.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } result, err := client.UploadBundle(context.Background(), UploadBundleOptions{ Root: root, IdempotencyKey: "producer.retry:one", }) if err != nil { t.Fatalf("UploadBundle() error = %v", err) } if result.RunID != "reports.20260604T120000Z.abcdef12" || result.Status != "accepted" { t.Fatalf("result = %#v, want accepted run", result) } } func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) { sourceRoot := t.TempDir() sourcePath := filepath.Join(sourceRoot, "producer-output.md") if err := os.WriteFile(sourcePath, []byte("producer data\n"), 0o600); err != nil { t.Fatalf("write source: %v", err) } tempDir := t.TempDir() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { entries := readArchiveEntries(t, r.Body) if got := string(entries["manifest.json"]); !strings.Contains(got, `"id": "reports.from.files"`) { t.Fatalf("manifest = %s, want uploaded id", got) } if got, want := string(entries["reports/report.md"]), "producer data\n"; got != want { t.Fatalf("uploaded file = %q, want %q", got, want) } if _, ok := entries["producer-output.md"]; ok { t.Fatal("archive used producer source path instead of bundle path") } writeAccepted(t, w, "reports.20260604T120000Z.abcdef12") })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } _, err = client.UploadFiles(context.Background(), UploadFilesOptions{ ID: "reports.from.files", Files: []sourcebundle.BundleFile{{ SourcePath: sourcePath, Path: "reports/report.md", }}, TempDir: tempDir, }) if err != nil { t.Fatalf("UploadFiles() error = %v", err) } if _, err := os.Stat(filepath.Join(sourceRoot, sourcebundle.ManifestName)); !os.IsNotExist(err) { t.Fatalf("producer source manifest stat = %v, want not exist", err) } entries, err := os.ReadDir(tempDir) if err != nil { t.Fatalf("read temp dir: %v", err) } if len(entries) != 0 { t.Fatalf("temp dir entries = %d, want cleanup", len(entries)) } } func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "original"}}) if err := os.WriteFile(filepath.Join(root, "report.md"), []byte("changed"), 0o600); err != nil { t.Fatalf("mutate bundle file: %v", err) } var requests atomic.Int64 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests.Add(1) t.Fatal("server should not receive request") })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err == nil { t.Fatal("UploadBundle() error = nil, want validation error") } if got := requests.Load(); got != 0 { t.Fatalf("requests = %d, want 0", got) } } func TestUploadBundleCanDisableLocalValidation(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "original"}}) if err := os.WriteFile(filepath.Join(root, "report.md"), []byte("changed"), 0o600); err != nil { t.Fatalf("mutate bundle file: %v", err) } var requests atomic.Int64 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests.Add(1) writeAccepted(t, w, "reports.20260604T120000Z.abcdef12") })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, DisableValidation: true}); err != nil { t.Fatalf("UploadBundle() error = %v", err) } if got := requests.Load(); got != 1 { t.Fatalf("requests = %d, want 1", got) } } func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) var attempts atomic.Int64 var keys []string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { keys = append(keys, r.Header.Get(idempotencyKeyHeader)) if attempts.Add(1) == 1 { writeJSONError(w, http.StatusServiceUnavailable, "busy", false) return } writeAccepted(t, w, "reports.20260604T120000Z.abcdef12") })) defer server.Close() client, err := NewClient(ClientOptions{ Endpoint: server.URL, Token: "secret", HTTPClient: server.Client(), Retry: RetryOptions{MaxAttempts: 2, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond}, }) if err != nil { t.Fatalf("NewClient() error = %v", err) } if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err != nil { t.Fatalf("UploadBundle() error = %v", err) } if got, want := attempts.Load(), int64(2); got != want { t.Fatalf("attempts = %d, want %d", got, want) } if len(keys) != 2 || keys[0] == "" || keys[0] != keys[1] { t.Fatalf("idempotency keys = %#v, want same generated key", keys) } if !regexp.MustCompile(`^[0-9a-f]{32}$`).MatchString(keys[0]) { t.Fatalf("generated key = %q, want 128-bit lowercase hex", keys[0]) } } func TestUploadResponseParsingAndNoRetryStatuses(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) tests := []struct { name string status int body string wantConflict bool wantMessage string wantRetryable bool }{ {name: "bad request", status: http.StatusBadRequest, body: `{"error":"bad bundle"}`, wantMessage: "bad bundle"}, {name: "unauthorized", status: http.StatusUnauthorized, body: `{"error":"bad token"}`, wantMessage: "bad token"}, {name: "conflict", status: http.StatusConflict, body: `{"error":"different manifest","retryable":true}`, wantConflict: true, wantMessage: "different manifest", wantRetryable: true}, {name: "too large", status: http.StatusRequestEntityTooLarge, body: `{"error":"too large"}`, wantMessage: "too large"}, {name: "unsupported", status: http.StatusUnsupportedMediaType, body: `{"error":"unsupported"}`, wantMessage: "unsupported"}, {name: "service unavailable", status: http.StatusServiceUnavailable, body: `{"error":"busy"}`, wantMessage: "busy"}, {name: "non json", status: http.StatusBadRequest, body: `plain failure`, wantMessage: "plain failure"}, {name: "unexpected", status: http.StatusTeapot, body: ``, wantMessage: "I'm a teapot"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { var attempts atomic.Int64 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { attempts.Add(1) w.WriteHeader(tt.status) _, _ = w.Write([]byte(tt.body)) })) defer server.Close() client, err := NewClient(ClientOptions{ Endpoint: server.URL, Token: "secret", HTTPClient: server.Client(), Retry: RetryOptions{MaxAttempts: 1}, }) if err != nil { t.Fatalf("NewClient() error = %v", err) } _, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"}) if err == nil { t.Fatal("UploadBundle() error = nil, want error") } var httpErr *HTTPError if !errors.As(err, &httpErr) { t.Fatalf("error = %T %v, want HTTPError", err, err) } if httpErr.StatusCode != tt.status || !strings.Contains(httpErr.Message, tt.wantMessage) || httpErr.Retryable != tt.wantRetryable { t.Fatalf("HTTPError = %#v, want status %d message %q retryable %t", httpErr, tt.status, tt.wantMessage, tt.wantRetryable) } var conflict *IdempotencyConflictError if got := errors.As(err, &conflict); got != tt.wantConflict { t.Fatalf("conflict error = %t, want %t", got, tt.wantConflict) } if got := attempts.Load(); got != 1 { t.Fatalf("attempts = %d, want 1", got) } }) } } func TestTokenRedactedFromHTTPError(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) token := "super-secret-token" server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { writeJSONError(w, http.StatusBadRequest, "token "+token+" rejected", false) })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: token, HTTPClient: server.Client(), Retry: RetryOptions{MaxAttempts: 1}}) if err != nil { t.Fatalf("NewClient() error = %v", err) } _, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"}) if err == nil { t.Fatal("UploadBundle() error = nil, want error") } if strings.Contains(err.Error(), token) { t.Fatalf("error exposed token: %v", err) } if !strings.Contains(err.Error(), redactedSecret) { t.Fatalf("error = %v, want redaction marker", err) } } func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) var attempts atomic.Int64 var keys []string client, err := NewClient(ClientOptions{ Endpoint: "http://upload.example", Token: "secret", HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) { keys = append(keys, request.Header.Get(idempotencyKeyHeader)) if attempts.Add(1) == 1 { return nil, temporaryNetworkError{} } return &http.Response{ StatusCode: http.StatusAccepted, Status: "202 Accepted", Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"run_id":"reports.20260604T120000Z.abcdef12","status":"accepted"}`)), Request: request, }, nil })}, Retry: RetryOptions{MaxAttempts: 2, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond}, }) if err != nil { t.Fatalf("NewClient() error = %v", err) } result, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "network-retry"}) if err != nil { t.Fatalf("UploadBundle() error = %v", err) } if result.RunID == "" { t.Fatalf("result = %#v, want run id", result) } if got, want := attempts.Load(), int64(2); got != want { t.Fatalf("attempts = %d, want %d", got, want) } if got, want := strings.Join(keys, ","), "network-retry,network-retry"; got != want { t.Fatalf("keys = %q, want %q", got, want) } } func TestContextCancellationDuringRetryBackoff(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) ctx, cancel := context.WithCancel(context.Background()) var attempts atomic.Int64 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { attempts.Add(1) cancel() writeJSONError(w, http.StatusServiceUnavailable, "busy", false) })) defer server.Close() client, err := NewClient(ClientOptions{ Endpoint: server.URL, Token: "secret", HTTPClient: server.Client(), Retry: RetryOptions{MaxAttempts: 2, BaseDelay: time.Hour, MaxDelay: time.Hour}, }) if err != nil { t.Fatalf("NewClient() error = %v", err) } _, err = client.UploadBundle(ctx, UploadBundleOptions{Root: root, IdempotencyKey: "cancel"}) if !errors.Is(err, context.Canceled) { t.Fatalf("UploadBundle() error = %v, want context.Canceled", err) } if got := attempts.Load(); got != 1 { t.Fatalf("attempts = %d, want 1", got) } } func TestStatusParsesRunStatusAndErrors(t *testing.T) { acceptedAt := time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if got, want := r.URL.Path, "/runs/reports.20260604T120000Z.abcdef12"; got != want { t.Fatalf("path = %q, want %q", got, want) } if got, want := r.Header.Get("Authorization"), "Bearer secret"; got != want { t.Fatalf("authorization = %q, want %q", got, want) } _ = json.NewEncoder(w).Encode(RunStatus{ RunID: "reports.20260604T120000Z.abcdef12", PipelineID: "reports", Status: "succeeded", AcceptedAt: acceptedAt, }) })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } status, err := client.Status(context.Background(), "reports.20260604T120000Z.abcdef12") if err != nil { t.Fatalf("Status() error = %v", err) } if status.RunID != "reports.20260604T120000Z.abcdef12" || status.PipelineID != "reports" || status.Status != "succeeded" || !status.AcceptedAt.Equal(acceptedAt) { t.Fatalf("status = %#v, want succeeded run", status) } errorServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { writeJSONError(w, http.StatusNotFound, "run not found", false) })) defer errorServer.Close() client, err = NewClient(ClientOptions{Endpoint: errorServer.URL, Token: "secret", HTTPClient: errorServer.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } _, err = client.Status(context.Background(), "missing") var httpErr *HTTPError if err == nil || !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusNotFound { t.Fatalf("Status() error = %v, want 404 HTTPError", err) } } func TestInvalidCallerIdempotencyKeyPreventsHTTPRequest(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) var requests atomic.Int64 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests.Add(1) })) defer server.Close() client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) if err != nil { t.Fatalf("NewClient() error = %v", err) } if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "bad key"}); err == nil { t.Fatal("UploadBundle() error = nil, want invalid key error") } if got := requests.Load(); got != 0 { t.Fatalf("requests = %d, want 0", got) } } type testFile struct { path string data string } func writeTestBundle(t *testing.T, id string, files []testFile) string { t.Helper() sourceRoot := t.TempDir() bundleFiles := make([]sourcebundle.BundleFile, 0, len(files)) for _, file := range files { sourcePath := filepath.Join(sourceRoot, filepath.FromSlash(file.path)) if err := os.MkdirAll(filepath.Dir(sourcePath), 0o755); err != nil { t.Fatalf("mkdir source parent: %v", err) } if err := os.WriteFile(sourcePath, []byte(file.data), 0o600); err != nil { t.Fatalf("write source file: %v", err) } bundleFiles = append(bundleFiles, sourcebundle.BundleFile{ SourcePath: sourcePath, Path: file.path, }) } root := filepath.Join(t.TempDir(), "bundle") if _, err := sourcebundle.WriteBundle(sourcebundle.WriteBundleOptions{ Root: root, ID: id, Created: time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC), Files: bundleFiles, }); err != nil { t.Fatalf("WriteBundle() error = %v", err) } return root } func readArchiveEntries(t *testing.T, body io.Reader) map[string][]byte { t.Helper() gzipReader, err := gzip.NewReader(body) if err != nil { t.Fatalf("open gzip archive: %v", err) } defer gzipReader.Close() tarReader := tar.NewReader(gzipReader) entries := map[string][]byte{} for { header, err := tarReader.Next() if errors.Is(err, io.EOF) { return entries } if err != nil { t.Fatalf("read tar archive: %v", err) } data, err := io.ReadAll(tarReader) if err != nil { t.Fatalf("read tar entry %q: %v", header.Name, err) } entries[header.Name] = data } } func entryNames(entries map[string][]byte) []string { ordered := []string{} for _, name := range []string{"manifest.json", "report.md", "nested/summary.txt", "reports/report.md", "unlisted.txt"} { if _, ok := entries[name]; ok { ordered = append(ordered, name) } } return ordered } func writeAccepted(t *testing.T, w http.ResponseWriter, runID string) { t.Helper() w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusAccepted) if err := json.NewEncoder(w).Encode(Result{RunID: runID, Status: "accepted"}); err != nil { t.Fatalf("write accepted response: %v", err) } } func writeJSONError(w http.ResponseWriter, status int, message string, retryable bool) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(map[string]any{"error": message, "retryable": retryable}) } type roundTripFunc func(*http.Request) (*http.Response, error) func (fn roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { return fn(request) } type temporaryNetworkError struct{} func (temporaryNetworkError) Error() string { return "temporary network failure" } func (temporaryNetworkError) Timeout() bool { return false } func (temporaryNetworkError) Temporary() bool { return true }