eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity
Files
main
distributor/pkg/upload/client_test.go

642 lines
24 KiB
Go
Raw Permalink Normal View History

Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if got, want := client.uploadURL("reports.daily"), "http://127.0.0.1:8080/base/v1/pipelines/reports.daily/upload"; got != want {
Add producer upload client package
2026-06-04 14:10:51 +00:00
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) {
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if got, want := r.URL.Path, "/v1/pipelines/reports.daily/upload"; got != want {
Add producer upload client package
2026-06-04 14:10:51 +00:00
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{
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
PipelineID: "reports.daily",
Add producer upload client package
2026-06-04 14:10:51 +00:00
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) {
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if got, want := r.URL.Path, "/v1/pipelines/reports.files/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
Add producer upload client package
2026-06-04 14:10:51 +00:00
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{
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
PipelineID: "reports.files",
ID: "reports.from.files",
Add producer upload client package
2026-06-04 14:10:51 +00:00
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))
}
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
func TestUploadMethodsRequirePipelineIDBeforeLocalWork(t *testing.T) {
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)
}
missingRoot := filepath.Join(t.TempDir(), "missing")
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: missingRoot}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") {
t.Fatalf("UploadBundle() error = %v, want missing pipeline id", err)
}
tempDir := t.TempDir()
sourcePath := filepath.Join(t.TempDir(), "report.md")
if err := os.WriteFile(sourcePath, []byte("data"), 0o600); err != nil {
t.Fatalf("write source: %v", err)
}
if _, err := client.UploadFiles(context.Background(), UploadFilesOptions{
ID: "reports.from.files",
Files: []sourcebundle.BundleFile{{
SourcePath: sourcePath,
Path: "report.md",
}},
TempDir: tempDir,
}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") {
t.Fatalf("UploadFiles() error = %v, want missing pipeline id", 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 no local bundle work", len(entries))
}
if got := requests.Load(); got != 0 {
t.Fatalf("requests = %d, want 0", got)
}
}
func TestUploadMethodsRejectInvalidPipelineIDBeforeHTTPRequest(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)
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)
}
for _, pipelineID := range []string{".reports", "reports/daily", "reports daily"} {
t.Run(pipelineID, func(t *testing.T) {
_, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: pipelineID, Root: root})
if err == nil || !strings.Contains(err.Error(), "pipeline id must be a slug-like identifier") {
t.Fatalf("UploadBundle() error = %v, want invalid pipeline id", err)
}
})
}
if got := requests.Load(); got != 0 {
t.Fatalf("requests = %d, want 0", got)
}
}
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err == nil {
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, DisableValidation: true}); err != nil {
Add producer upload client package
2026-06-04 14:10:51 +00:00
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) {
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if got, want := r.URL.Path, "/v1/pipelines/reports/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err != nil {
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
_, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"})
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
_, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"})
Add producer upload client package
2026-06-04 14:10:51 +00:00
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))
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if got, want := request.URL.Path, "/v1/pipelines/reports/upload"; got != want {
t.Fatalf("path = %q, want %q", got, want)
}
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
result, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "network-retry"})
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
_, err = client.UploadBundle(ctx, UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "cancel"})
Add producer upload client package
2026-06-04 14:10:51 +00:00
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)
}
Route upload client by pipeline
2026-06-08 04:37:33 +00:00
if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "bad key"}); err == nil {
Add producer upload client package
2026-06-04 14:10:51 +00:00
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
}
Reference in New Issue Copy Permalink