package distributor import ( "context" "encoding/json" "errors" "fmt" "strings" "testing" "time" distributorupload "gitea.maximumdirect.net/eric/distributor/pkg/upload" "gitea.maximumdirect.net/eric/weatherreporter/internal/config" ) func TestUploadUsesConfiguredClientAndFiles(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" cfg.TokenEnv = "DISTRIBUTOR_UPLOAD_TOKEN" cfg.Timeout = 15 * time.Second t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{ result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, status: runStatus{RunID: "run-123", PipelineID: "reports", Status: "succeeded", Report: json.RawMessage(`{"actions":[{"action":"replace_older"}]}`)}, }, } client := newClient(cfg, factory.newClient) result, err := client.Upload(context.Background(), UploadRequest{ PipelineID: "weatherreporter.daily", BundleID: "weatherreporter.home.daily.run", IdempotencyKey: "weatherreporter.home.daily.run", Files: []UploadFile{ {SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/report.md"}, {SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/latest.md"}, }, CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC), }) if err != nil { t.Fatalf("Upload() error = %v", err) } if result.RunID != "run-123" || result.Status != "succeeded" || result.UploadStatus != "accepted" { t.Fatalf("result = %#v, want accepted run", result) } if result.RunStatus == nil || result.RunStatus.PipelineID != "reports" || !strings.Contains(string(result.RunStatus.Report), "replace_older") { t.Fatalf("RunStatus = %#v, want parsed run report", result.RunStatus) } if factory.endpoint != cfg.Endpoint { t.Fatalf("factory endpoint = %q, want %q", factory.endpoint, cfg.Endpoint) } if factory.token != "secret-token" { t.Fatalf("factory token = %q, want secret-token", factory.token) } if factory.timeout != 15*time.Second { t.Fatalf("factory timeout = %s, want 15s", factory.timeout) } got := factory.client.opts if got.PipelineID != "weatherreporter.daily" { t.Fatalf("PipelineID = %q, want weatherreporter.daily", got.PipelineID) } if got.BundleID != "weatherreporter.home.daily.run" { t.Fatalf("BundleID = %q, want weatherreporter.home.daily.run", got.BundleID) } if got.IdempotencyKey != "weatherreporter.home.daily.run" { t.Fatalf("IdempotencyKey = %q, want weatherreporter.home.daily.run", got.IdempotencyKey) } if len(got.Files) != 2 { t.Fatalf("files = %#v, want two mappings", got.Files) } if got.Files[0].SourcePath != "/tmp/report.md" || got.Files[0].BundlePath != "2026-06-07/daily/report.md" { t.Fatalf("first file = %#v, want archive mapping", got.Files[0]) } if got.Files[1].SourcePath != "/tmp/report.md" || got.Files[1].BundlePath != "2026-06-07/daily/latest.md" { t.Fatalf("second file = %#v, want latest mapping", got.Files[1]) } if got.CreatedAt.IsZero() { t.Fatal("CreatedAt is zero, want generated report timestamp") } if factory.client.statusRunID != "run-123" { t.Fatalf("Status runID = %q, want run-123", factory.client.statusRunID) } } func TestUploadRejectsMissingInputs(t *testing.T) { cfg := config.Defaults().Notify.Distributor t.Setenv(cfg.TokenEnv, "secret-token") tests := []struct { name string mutate func(*Client, *UploadRequest) wantErr string }{ { name: "Token", mutate: func(c *Client, req *UploadRequest) { t.Setenv(c.TokenEnv, "") }, wantErr: "token environment variable", }, { name: "PipelineID", mutate: func(c *Client, req *UploadRequest) { req.PipelineID = "" }, wantErr: "pipeline id is required", }, { name: "Files", mutate: func(c *Client, req *UploadRequest) { req.Files = nil }, wantErr: "upload files are required", }, { name: "SourcePath", mutate: func(c *Client, req *UploadRequest) { req.Files[0].SourcePath = "" }, wantErr: "source path is required", }, { name: "BundlePath", mutate: func(c *Client, req *UploadRequest) { req.Files[0].BundlePath = "" }, wantErr: "bundle path is required", }, { name: "UploadClientFactory", mutate: func(c *Client, req *UploadRequest) { c.newUploadClient = nil }, wantErr: "upload client factory is required", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Setenv(cfg.TokenEnv, "secret-token") client := newClient(cfg, (&fakeUploadFactory{client: &fakeUploadClient{}}).newClient) req := validUploadRequest() tt.mutate(client, &req) _, err := client.Upload(context.Background(), req) if err == nil { t.Fatal("Upload() error = nil, want error") } if !strings.Contains(err.Error(), tt.wantErr) { t.Fatalf("error = %q, want %q", err.Error(), tt.wantErr) } if strings.Contains(err.Error(), "secret-token") { t.Fatalf("error = %q, want no token value", err.Error()) } }) } } func TestUploadWrapsFactoryErrorWithoutToken(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ err: fmt.Errorf("factory failed with secret-token"), } client := newClient(cfg, factory.newClient) _, err := client.Upload(context.Background(), validUploadRequest()) if err == nil { t.Fatal("Upload() error = nil, want error") } if strings.Contains(err.Error(), "secret-token") { t.Fatalf("error = %q, want no token value", err.Error()) } if !strings.Contains(err.Error(), cfg.Endpoint) { t.Fatalf("error = %q, want endpoint context", err.Error()) } } func TestUploadWrapsUploadFailureWithContextWithoutToken(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{err: fmt.Errorf("server rejected secret-token")}, } client := newClient(cfg, factory.newClient) req := validUploadRequest() _, err := client.Upload(context.Background(), req) if err == nil { t.Fatal("Upload() error = nil, want error") } for _, want := range []string{cfg.Endpoint, req.PipelineID, req.BundleID, req.IdempotencyKey, req.Files[0].SourcePath, req.Files[0].BundlePath, req.Files[1].BundlePath} { if !strings.Contains(err.Error(), want) { t.Fatalf("error = %q, want context %q", err.Error(), want) } } if strings.Contains(err.Error(), "secret-token") { t.Fatalf("error = %q, want no token value", err.Error()) } } func TestUploadReturnsAcceptedWhenStatusLookupFails(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{ result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, statusErr: fmt.Errorf("status rejected secret-token"), }, } client := newClient(cfg, factory.newClient) result, err := client.Upload(context.Background(), validUploadRequest()) if err != nil { t.Fatalf("Upload() error = %v, want accepted upload despite status lookup failure", err) } if result.Status != "accepted" || result.StatusError == "" { t.Fatalf("result = %#v, want accepted status with status error", result) } if strings.Contains(result.StatusError, "secret-token") { t.Fatalf("StatusError = %q, want token redacted", result.StatusError) } } func TestUploadPollsUntilTerminalStatus(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" cfg.Timeout = 2 * time.Second t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{ result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, statuses: []runStatus{ {RunID: "run-123", Status: "accepted"}, {RunID: "run-123", Status: "succeeded", Report: json.RawMessage(`{"actions":[{"action":"replace_older"}]}`)}, }, }, } client := newClient(cfg, factory.newClient) result, err := client.Upload(context.Background(), validUploadRequest()) if err != nil { t.Fatalf("Upload() error = %v", err) } if result.Status != "succeeded" || result.RunStatus == nil || !strings.Contains(string(result.RunStatus.Report), "replace_older") { t.Fatalf("result = %#v, want terminal succeeded status with run report", result) } if factory.client.statusCalls != 2 { t.Fatalf("status calls = %d, want 2", factory.client.statusCalls) } } func TestUploadReturnsLatestStatusWhenPollingTimesOut(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" cfg.Timeout = time.Millisecond t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{ result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, status: runStatus{RunID: "run-123", Status: "running"}, }, } client := newClient(cfg, factory.newClient) result, err := client.Upload(context.Background(), validUploadRequest()) if err != nil { t.Fatalf("Upload() error = %v, want accepted upload with status timeout recorded", err) } if result.Status != "running" || result.StatusError == "" { t.Fatalf("result = %#v, want latest status and status timeout", result) } } func TestUploadFailsWhenDistributorRunFailed(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{ result: uploadFilesResult{RunID: "run-123", Status: "accepted"}, status: runStatus{ RunID: "run-123", Status: "failed", Error: "destination rejected secret-token", Report: json.RawMessage(`{"actions":[{"action":"failed"}]}`), }, }, } client := newClient(cfg, factory.newClient) result, err := client.Upload(context.Background(), validUploadRequest()) if err == nil { t.Fatal("Upload() error = nil, want failed distributor run error") } if result.RunStatus == nil || result.RunStatus.Status != "failed" || !strings.Contains(string(result.RunStatus.Report), "failed") { t.Fatalf("result = %#v, want failed run status report", result) } if strings.Contains(err.Error(), "secret-token") || strings.Contains(result.RunStatus.Error, "secret-token") { t.Fatalf("error/result leaked token: err=%q result=%#v", err.Error(), result) } } func TestUploadPreservesIdempotencyConflictDiagnosis(t *testing.T) { cfg := config.Defaults().Notify.Distributor cfg.Endpoint = "https://distributor.example.test" t.Setenv(cfg.TokenEnv, "secret-token") factory := &fakeUploadFactory{ client: &fakeUploadClient{ err: &distributorupload.IdempotencyConflictError{ HTTPError: distributorupload.HTTPError{ StatusCode: 409, Status: "409 Conflict", Message: "conflicting upload for secret-token", }, }, }, } client := newClient(cfg, factory.newClient) _, err := client.Upload(context.Background(), validUploadRequest()) if err == nil { t.Fatal("Upload() error = nil, want error") } var conflict *IdempotencyConflictError if !errors.As(err, &conflict) { t.Fatalf("Upload() error = %T %v, want IdempotencyConflictError", err, err) } if !strings.Contains(err.Error(), "idempotency conflict") { t.Fatalf("error = %q, want idempotency conflict diagnosis", err.Error()) } if strings.Contains(err.Error(), "secret-token") { t.Fatalf("error = %q, want no token value", err.Error()) } } func validUploadRequest() UploadRequest { return UploadRequest{ PipelineID: "weatherreporter.daily", BundleID: "weatherreporter.home.daily.run", IdempotencyKey: "weatherreporter.home.daily.run", Files: []UploadFile{ {SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/report.md"}, {SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/latest.md"}, }, CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC), } } type fakeUploadFactory struct { endpoint string token string timeout time.Duration client *fakeUploadClient err error } func (f *fakeUploadFactory) newClient(endpoint, token string, timeout time.Duration) (uploadClient, error) { f.endpoint = endpoint f.token = token f.timeout = timeout if f.err != nil { return nil, f.err } return f.client, nil } type fakeUploadClient struct { opts uploadFilesOptions statusRunID string statusCalls int result uploadFilesResult status runStatus statuses []runStatus err error statusErr error } func (c *fakeUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) { c.opts = opts if c.err != nil { return uploadFilesResult{}, c.err } return c.result, nil } func (c *fakeUploadClient) Status(ctx context.Context, runID string) (runStatus, error) { c.statusRunID = runID c.statusCalls++ if c.statusErr != nil { return runStatus{}, c.statusErr } if len(c.statuses) > 0 { index := c.statusCalls - 1 if index >= len(c.statuses) { index = len(c.statuses) - 1 } return c.statuses[index], nil } return c.status, nil }