Add single-pipeline run entrypoint

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) {

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")

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, &notFound)
+}
+
 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)