From 761a2f0bc20293bd98c40ed8c7fdc8331b873c1f Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 3 Jun 2026 11:46:44 +0000 Subject: [PATCH] Add single-pipeline run entrypoint --- internal/app/run.go | 43 +++++++++++++++ internal/app/run_test.go | 100 ++++++++++++++++++++++++++++++++++ internal/app/source_select.go | 16 +++++- 3 files changed, 158 insertions(+), 1 deletion(-) diff --git a/internal/app/run.go b/internal/app/run.go index 8edc2b2..63389ba 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -21,6 +21,14 @@ type RunOptions struct { Notifier notify.Notifier } +type RunPipelineOptions struct { + ConfigPath string + PipelineID string + DryRun bool + Force bool + Notifier notify.Notifier +} + func Run(ctx context.Context, options RunOptions) error { if err := ValidateOutputFormat(options.OutputFormat); err != nil { return err @@ -40,12 +48,47 @@ func Run(ctx context.Context, options RunOptions) error { return runConfig(ctx, cfg, options) } +func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, error) { + if err := ctx.Err(); err != nil { + return RunReport{}, err + } + + configPath := options.ConfigPath + if configPath == "" { + configPath = config.DefaultConfigPath + } + cfg, err := config.LoadFile(configPath) + if err != nil { + return RunReport{}, err + } + return runPipelineConfig(ctx, cfg, options) +} + func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error { return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) } type backendFactoryProvider func(config.Environment) *backendFactory +func runPipelineConfig(ctx context.Context, cfg config.Config, options RunPipelineOptions) (RunReport, error) { + return runPipelineConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) +} + +func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) { + pipeline, ok := findPipeline(cfg, options.PipelineID) + if !ok { + return RunReport{}, PipelineNotFoundError{ID: options.PipelineID} + } + return buildRunReportWithBackendFactory(ctx, config.Config{ + Secrets: cfg.Secrets, + Pipelines: []config.Pipeline{pipeline}, + }, RunOptions{ + DryRun: options.DryRun, + Force: options.Force, + Notifier: options.Notifier, + }, provider) +} + func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error { report, err := buildRunReportWithBackendFactory(ctx, cfg, options, provider) if err != nil && !IsPartialResultError(err) { diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 5caa740..a87c108 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -811,6 +811,83 @@ func TestBuildRunReportIncludesPartialFailures(t *testing.T) { } } +func TestRunPipelineRunsOnlyRequestedPipeline(t *testing.T) { + firstSource := t.TempDir() + secondSource := t.TempDir() + firstDestination := t.TempDir() + secondDestination := t.TempDir() + writeSourceBundle(t, firstSource, "", testBundleOptions{ID: "reports.one"}) + writeSourceBundle(t, secondSource, "", testBundleOptions{ID: "reports.two"}) + configPath := writeTwoPipelineConfig(t, firstSource, firstDestination, secondSource, secondDestination) + notifier := &recordingNotifier{} + + report, err := RunPipeline(context.Background(), RunPipelineOptions{ + ConfigPath: configPath, + PipelineID: "reports-one", + Notifier: notifier, + }) + if err != nil { + t.Fatalf("RunPipeline() error = %v", err) + } + if got, want := len(report.Pipelines), 1; got != want { + t.Fatalf("pipeline count = %d, want %d", got, want) + } + if report.Pipelines[0].ID != "reports-one" { + t.Fatalf("pipeline id = %q, want reports-one", report.Pipelines[0].ID) + } + if got, want := len(report.Actions), 1; got != want { + t.Fatalf("action count = %d, want %d", got, want) + } + if report.Actions[0].PipelineID != "reports-one" || report.Actions[0].Action != "publish_new" { + t.Fatalf("action = %#v, want reports-one publish_new", report.Actions[0]) + } + if got, want := len(notifier.events), 1; got != want { + t.Fatalf("notification count = %d, want %d", got, want) + } + if notifier.events[0].PipelineID != "reports-one" { + t.Fatalf("notification pipeline = %q, want reports-one", notifier.events[0].PipelineID) + } + testutil.AssertFile(t, filepath.Join(firstDestination, "report.md"), "# Report\nSunny.\n") + if entries, err := os.ReadDir(secondDestination); err != nil || len(entries) != 0 { + t.Fatalf("second destination entries = %v err=%v, want empty", entries, err) + } +} + +func TestRunPipelineUnknownIDReturnsNotFound(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + + _, err := RunPipeline(context.Background(), RunPipelineOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + PipelineID: "missing", + }) + if err == nil || !IsPipelineNotFound(err) { + t.Fatalf("RunPipeline() error = %v, want pipeline not found", err) + } + if !strings.Contains(err.Error(), `pipeline "missing" not found`) { + t.Fatalf("RunPipeline() error = %v, want pipeline id in message", err) + } +} + +func TestRunStillRunsAllConfiguredPipelines(t *testing.T) { + firstSource := t.TempDir() + secondSource := t.TempDir() + firstDestination := t.TempDir() + secondDestination := t.TempDir() + writeSourceBundle(t, firstSource, "", testBundleOptions{ID: "reports.one"}) + writeSourceBundle(t, secondSource, "", testBundleOptions{ID: "reports.two"}) + + err := Run(context.Background(), RunOptions{ + ConfigPath: writeTwoPipelineConfig(t, firstSource, firstDestination, secondSource, secondDestination), + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + testutil.AssertFile(t, filepath.Join(firstDestination, "report.md"), "# Report\nSunny.\n") + testutil.AssertFile(t, filepath.Join(secondDestination, "report.md"), "# Report\nSunny.\n") +} + func TestRunDoesNotNotifyForSkippedDestination(t *testing.T) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() @@ -1439,6 +1516,29 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination) } +func writeTwoPipelineConfig(t *testing.T, firstSource, firstDestination, secondSource, secondDestination string) string { + t.Helper() + return writeConfigFile(t, ` +pipelines: + - id: reports-one + source: + backend: local + path: `+firstSource+` + destinations: + - id: archive + backend: local + path: `+firstDestination+` + - id: reports-two + source: + backend: local + path: `+secondSource+` + destinations: + - id: archive + backend: local + path: `+secondDestination+` +`) +} + func writeConfigFile(t *testing.T, body string) string { t.Helper() path := filepath.Join(t.TempDir(), "config.yml") diff --git a/internal/app/source_select.go b/internal/app/source_select.go index aa32af9..c107597 100644 --- a/internal/app/source_select.go +++ b/internal/app/source_select.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "fmt" "gitea.maximumdirect.net/eric/distributor/internal/bundle" @@ -9,6 +10,19 @@ import ( "gitea.maximumdirect.net/eric/distributor/internal/storage" ) +type PipelineNotFoundError struct { + ID string +} + +func (e PipelineNotFoundError) Error() string { + return fmt.Sprintf("pipeline %q not found", e.ID) +} + +func IsPipelineNotFound(err error) bool { + var notFound PipelineNotFoundError + return errors.As(err, ¬Found) +} + type sourceCommandOptions struct { CommandName string Path string @@ -70,7 +84,7 @@ func selectSourceBundlesFromConfig(ctx context.Context, cfg config.Config, optio } pipeline, ok := findPipeline(cfg, options.PipelineID) if !ok { - return sourceSelection{}, fmt.Errorf("pipeline %q not found", options.PipelineID) + return sourceSelection{}, PipelineNotFoundError{ID: options.PipelineID} } backends := provider(secretLoad.Environment) sourceBackend, err := backends.openSource(ctx, pipeline.Source)