356 lines
11 KiB
Go
356 lines
11 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/distributor/internal/config"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/ingest"
|
|
)
|
|
|
|
type fakeUploadCoordinator struct {
|
|
canAccept bool
|
|
submit func(context.Context, UploadRequest) (UploadRunRecord, error)
|
|
status func(UploadRunID) (UploadRunRecord, bool)
|
|
}
|
|
|
|
func (fake fakeUploadCoordinator) CanAccept() bool {
|
|
return fake.canAccept
|
|
}
|
|
|
|
func (fake fakeUploadCoordinator) Submit(ctx context.Context, request UploadRequest) (UploadRunRecord, error) {
|
|
if fake.submit == nil {
|
|
return UploadRunRecord{}, errors.New("unexpected submit")
|
|
}
|
|
return fake.submit(ctx, request)
|
|
}
|
|
|
|
func (fake fakeUploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) {
|
|
if fake.status == nil {
|
|
return UploadRunRecord{}, false
|
|
}
|
|
return fake.status(runID)
|
|
}
|
|
|
|
func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
|
|
cfg := uploadHTTPTestConfig()
|
|
|
|
_, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) {
|
|
return "", false
|
|
}))
|
|
if err == nil || !strings.Contains(err.Error(), "UPLOAD_TOKEN") {
|
|
t.Fatalf("resolveUploadTokens() error = %v, want missing UPLOAD_TOKEN", err)
|
|
}
|
|
|
|
cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{
|
|
ID: "weekly",
|
|
Source: config.Backend{
|
|
Backend: config.BackendHTTPUpload,
|
|
Upload: config.HTTPUpload{TokenEnv: "OTHER_UPLOAD_TOKEN"},
|
|
},
|
|
Destinations: cfg.Pipelines[0].Destinations,
|
|
})
|
|
config.ApplyDefaults(&cfg)
|
|
secret := "super-secret-token"
|
|
_, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
|
|
"UPLOAD_TOKEN": secret,
|
|
"OTHER_UPLOAD_TOKEN": secret,
|
|
}))
|
|
if err == nil {
|
|
t.Fatal("resolveUploadTokens() error = nil, want duplicate token error")
|
|
}
|
|
if strings.Contains(err.Error(), secret) {
|
|
t.Fatalf("duplicate token error exposed secret value: %q", err)
|
|
}
|
|
}
|
|
|
|
func TestNewUploadHTTPHandlerAcceptsDefaultedConfig(t *testing.T) {
|
|
cfg := uploadHTTPTestConfig()
|
|
cfg.Server.HTTP.Bind = ""
|
|
cfg.Server.HTTP.StagingRoot = ""
|
|
cfg.Server.HTTP.MaxUploadSize = nil
|
|
cfg.Server.HTTP.QueueSize = 0
|
|
cfg.Server.HTTP.MaxConcurrency = 0
|
|
cfg.Server.HTTP.Retention = nil
|
|
cfg.Pipelines[0].Source.Upload.StagingPath = ""
|
|
cfg.Pipelines[0].Source.Upload.MaxUploadSize = nil
|
|
|
|
handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
|
|
"UPLOAD_TOKEN": "secret",
|
|
}))
|
|
if err != nil {
|
|
t.Fatalf("newUploadHTTPHandler() error = %v", err)
|
|
}
|
|
if handler == nil {
|
|
t.Fatal("newUploadHTTPHandler() = nil")
|
|
}
|
|
}
|
|
|
|
func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
|
|
var submitted UploadRequest
|
|
handler := uploadHTTPHandler{
|
|
coordinator: fakeUploadCoordinator{
|
|
canAccept: true,
|
|
submit: func(_ context.Context, request UploadRequest) (UploadRunRecord, error) {
|
|
submitted = request
|
|
body, err := io.ReadAll(request.Body)
|
|
if err != nil {
|
|
t.Fatalf("read submitted body: %v", err)
|
|
}
|
|
if string(body) != "archive" {
|
|
t.Fatalf("submitted body = %q, want archive", body)
|
|
}
|
|
return UploadRunRecord{ID: "reports.20260603T120000Z.abcdef12", Status: UploadStatusAccepted}, nil
|
|
},
|
|
},
|
|
tokens: map[string]string{"valid-token": "reports"},
|
|
}
|
|
recorder := httptest.NewRecorder()
|
|
request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
|
|
request.Header.Set("Authorization", "Bearer valid-token")
|
|
request.Header.Set("Content-Type", "application/x-tar")
|
|
|
|
handler.ServeHTTP(recorder, request)
|
|
|
|
if recorder.Code != http.StatusAccepted {
|
|
t.Fatalf("status = %d, want %d; body = %q", recorder.Code, http.StatusAccepted, recorder.Body.String())
|
|
}
|
|
if submitted.PipelineID != "reports" {
|
|
t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID)
|
|
}
|
|
var response uploadAcceptedResponse
|
|
if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if response.RunID != "reports.20260603T120000Z.abcdef12" || response.Status != UploadStatusAccepted {
|
|
t.Fatalf("response = %#v, want accepted run id", response)
|
|
}
|
|
if strings.Contains(recorder.Body.String(), "valid-token") {
|
|
t.Fatalf("response exposed token: %q", recorder.Body.String())
|
|
}
|
|
}
|
|
|
|
func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
|
|
handler := uploadHTTPHandler{
|
|
coordinator: fakeUploadCoordinator{canAccept: true},
|
|
tokens: map[string]string{"valid-token": "reports"},
|
|
}
|
|
|
|
for _, authHeader := range []string{"", "Bearer wrong-token"} {
|
|
recorder := httptest.NewRecorder()
|
|
request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
|
|
request.Header.Set("Authorization", authHeader)
|
|
request.Header.Set("Content-Type", "application/x-tar")
|
|
|
|
handler.ServeHTTP(recorder, request)
|
|
|
|
if recorder.Code != http.StatusUnauthorized {
|
|
t.Fatalf("auth %q status = %d, want %d", authHeader, recorder.Code, http.StatusUnauthorized)
|
|
}
|
|
if strings.Contains(recorder.Body.String(), "valid-token") || strings.Contains(recorder.Body.String(), "wrong-token") {
|
|
t.Fatalf("unauthorized response exposed token: %q", recorder.Body.String())
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
canAccept bool
|
|
url string
|
|
contentType string
|
|
body io.Reader
|
|
wantStatus int
|
|
}{
|
|
{
|
|
name: "unsupported content type",
|
|
canAccept: true,
|
|
url: "/upload",
|
|
contentType: "application/zip",
|
|
body: strings.NewReader("archive"),
|
|
wantStatus: http.StatusUnsupportedMediaType,
|
|
},
|
|
{
|
|
name: "full queue",
|
|
canAccept: false,
|
|
url: "/upload",
|
|
contentType: "application/x-tar",
|
|
body: &countingReader{reader: strings.NewReader("archive")},
|
|
wantStatus: http.StatusServiceUnavailable,
|
|
},
|
|
{
|
|
name: "submitted pipeline id",
|
|
canAccept: true,
|
|
url: "/upload?pipeline_id=reports",
|
|
contentType: "application/x-tar",
|
|
body: strings.NewReader("archive"),
|
|
wantStatus: http.StatusBadRequest,
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
handler := uploadHTTPHandler{
|
|
coordinator: fakeUploadCoordinator{
|
|
canAccept: tt.canAccept,
|
|
submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
|
|
t.Fatal("Submit should not be called")
|
|
return UploadRunRecord{}, nil
|
|
},
|
|
},
|
|
tokens: map[string]string{"valid-token": "reports"},
|
|
}
|
|
recorder := httptest.NewRecorder()
|
|
request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
|
|
request.Header.Set("Authorization", "Bearer valid-token")
|
|
request.Header.Set("Content-Type", tt.contentType)
|
|
|
|
handler.ServeHTTP(recorder, request)
|
|
|
|
if recorder.Code != tt.wantStatus {
|
|
t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
|
|
}
|
|
if reader, ok := tt.body.(*countingReader); ok && reader.reads != 0 {
|
|
t.Fatalf("full queue read body %d time(s), want zero", reader.reads)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
err error
|
|
wantStatus int
|
|
}{
|
|
{name: "oversized", err: ingest.ErrUploadTooLarge, wantStatus: http.StatusRequestEntityTooLarge},
|
|
{name: "unsupported", err: ingest.ErrUnsupportedContentType, wantStatus: http.StatusUnsupportedMediaType},
|
|
{name: "malformed", err: errors.New("malformed archive"), wantStatus: http.StatusBadRequest},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
handler := uploadHTTPHandler{
|
|
coordinator: fakeUploadCoordinator{
|
|
canAccept: true,
|
|
submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
|
|
return UploadRunRecord{}, tt.err
|
|
},
|
|
},
|
|
tokens: map[string]string{"valid-token": "reports"},
|
|
}
|
|
recorder := httptest.NewRecorder()
|
|
request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
|
|
request.Header.Set("Authorization", "Bearer valid-token")
|
|
request.Header.Set("Content-Type", "application/x-tar")
|
|
|
|
handler.ServeHTTP(recorder, request)
|
|
|
|
if recorder.Code != tt.wantStatus {
|
|
t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
|
|
finishedAt := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)
|
|
handler := uploadHTTPHandler{
|
|
coordinator: fakeUploadCoordinator{
|
|
canAccept: true,
|
|
status: func(runID UploadRunID) (UploadRunRecord, bool) {
|
|
if runID != "reports.20260603T120000Z.abcdef12" {
|
|
return UploadRunRecord{}, false
|
|
}
|
|
return UploadRunRecord{
|
|
ID: runID,
|
|
PipelineID: "reports",
|
|
Status: UploadStatusSucceeded,
|
|
FinishedAt: &finishedAt,
|
|
}, true
|
|
},
|
|
},
|
|
tokens: map[string]string{"valid-token": "reports"},
|
|
}
|
|
|
|
recorder := httptest.NewRecorder()
|
|
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/healthz", nil))
|
|
if recorder.Code != http.StatusOK {
|
|
t.Fatalf("health status = %d, want %d", recorder.Code, http.StatusOK)
|
|
}
|
|
|
|
recorder = httptest.NewRecorder()
|
|
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/runs/reports.20260603T120000Z.abcdef12", nil))
|
|
if recorder.Code != http.StatusOK {
|
|
t.Fatalf("run status = %d, want %d; body = %q", recorder.Code, http.StatusOK, recorder.Body.String())
|
|
}
|
|
var record UploadRunRecord
|
|
if err := json.Unmarshal(recorder.Body.Bytes(), &record); err != nil {
|
|
t.Fatalf("decode run status: %v", err)
|
|
}
|
|
if record.ID != "reports.20260603T120000Z.abcdef12" || record.Status != UploadStatusSucceeded {
|
|
t.Fatalf("record = %#v, want succeeded run status", record)
|
|
}
|
|
|
|
recorder = httptest.NewRecorder()
|
|
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/runs/unknown", nil))
|
|
if recorder.Code != http.StatusNotFound {
|
|
t.Fatalf("unknown run status = %d, want %d", recorder.Code, http.StatusNotFound)
|
|
}
|
|
}
|
|
|
|
type countingReader struct {
|
|
reader io.Reader
|
|
reads int
|
|
}
|
|
|
|
func (reader *countingReader) Read(data []byte) (int, error) {
|
|
reader.reads++
|
|
return reader.reader.Read(data)
|
|
}
|
|
|
|
func uploadHTTPTestConfig() config.Config {
|
|
size := config.ByteSize(1024)
|
|
retention := config.Duration(24 * time.Hour)
|
|
cfg := config.Config{
|
|
Server: config.Server{HTTP: config.HTTPServer{
|
|
Bind: config.DefaultHTTPBind,
|
|
StagingRoot: "/tmp/distributor-test",
|
|
MaxUploadSize: &size,
|
|
QueueSize: 2,
|
|
MaxConcurrency: 1,
|
|
Retention: &retention,
|
|
}},
|
|
Pipelines: []config.Pipeline{{
|
|
ID: "reports",
|
|
Source: config.Backend{
|
|
Backend: config.BackendHTTPUpload,
|
|
Upload: config.HTTPUpload{
|
|
TokenEnv: "UPLOAD_TOKEN",
|
|
StagingPath: "/tmp/distributor-test/reports",
|
|
MaxUploadSize: &size,
|
|
},
|
|
},
|
|
Destinations: []config.Destination{{
|
|
ID: "local",
|
|
Backend: config.BackendLocal,
|
|
Path: "/tmp/distributor-output",
|
|
Publish: &config.PublishPolicy{Source: true},
|
|
}},
|
|
}},
|
|
}
|
|
config.ApplyDefaults(&cfg)
|
|
return cfg
|
|
}
|
|
|
|
func uploadHTTPTestEnvironment(values map[string]string) config.Environment {
|
|
return config.NewEnvironment(values, func(string) (string, bool) {
|
|
return "", false
|
|
})
|
|
}
|