642 lines
24 KiB
Go
642 lines
24 KiB
Go
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("reports.daily"), "http://127.0.0.1:8080/base/v1/pipelines/reports.daily/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, "/v1/pipelines/reports.daily/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{
|
|
PipelineID: "reports.daily",
|
|
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) {
|
|
if got, want := r.URL.Path, "/v1/pipelines/reports.files/upload"; got != want {
|
|
t.Fatalf("path = %q, want %q", got, want)
|
|
}
|
|
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{
|
|
PipelineID: "reports.files",
|
|
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 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)
|
|
}
|
|
}
|
|
|
|
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{PipelineID: "reports", 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{PipelineID: "reports", 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) {
|
|
if got, want := r.URL.Path, "/v1/pipelines/reports/upload"; got != want {
|
|
t.Fatalf("path = %q, want %q", got, want)
|
|
}
|
|
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{PipelineID: "reports", 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{PipelineID: "reports", 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{PipelineID: "reports", 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 got, want := request.URL.Path, "/v1/pipelines/reports/upload"; got != want {
|
|
t.Fatalf("path = %q, want %q", got, want)
|
|
}
|
|
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{PipelineID: "reports", 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{PipelineID: "reports", 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{PipelineID: "reports", 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
|
|
}
|