distributor/internal/app/run_test.go

package app

import (
	"bytes"
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
	"gitea.maximumdirect.net/eric/distributor/internal/config"
	"gitea.maximumdirect.net/eric/distributor/internal/notify"
	"gitea.maximumdirect.net/eric/distributor/internal/state"
	"gitea.maximumdirect.net/eric/distributor/internal/storage"
	"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)

func TestRunDryRunPrintsConfigSummary(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	configPath := writeLocalConfig(t, sourceRoot, destinationRoot)

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: configPath,
		DryRun:     true,
		Stdout:     &stdout,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}

	output := stdout.String()
	for _, want := range []string{
		"Configured pipelines: 1",
		"- pipeline=reports source=local bundles=1 destinations=archive",
		"bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt",
		"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
	} {
		if !strings.Contains(output, want) {
			t.Fatalf("Run() output = %q, want substring %q", output, want)
		}
	}
}

func TestRunDryRunUsesReadOnlySSHKnownHosts(t *testing.T) {
	sourceRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	cfg := config.Config{Pipelines: []config.Pipeline{{
		ID:     "reports",
		Source: config.Backend{Backend: config.BackendLocal, Path: sourceRoot},
		Destinations: []config.Destination{{
			ID:      "archive",
			Backend: config.BackendSSH,
			Host:    "destination.example.com",
			Path:    "/archive",
		}},
	}}}
	config.ApplyDefaults(&cfg)
	var got storage.OpenConfig
	provider := func(environment config.Environment) *backendFactory {
		registry := storage.NewRegistry()
		if err := registry.Register(config.BackendLocal, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			return local.New(cfg[storagePathKey])
		}); err != nil {
			t.Fatalf("register local backend: %v", err)
		}
		if err := registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			got = cfg
			return fake.New(), nil
		}); err != nil {
			t.Fatalf("register ssh backend: %v", err)
		}
		return &backendFactory{registry: registry, environment: environment}
	}

	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true}, provider); err != nil {
		t.Fatalf("runConfigWithBackendFactory() error = %v", err)
	}
	if got[sshReadOnlyHostsKey] != "true" {
		t.Fatalf("open config %s = %q, want true", sshReadOnlyHostsKey, got[sshReadOnlyHostsKey])
	}
}

func TestRunLoadsSecretsBeforeOpeningBackends(t *testing.T) {
	sourceRoot := filepath.Join(t.TempDir(), "missing-source")
	destinationRoot := t.TempDir()
	configPath := writeConfigFile(t, `
secrets:
  directory: `+filepath.Join(t.TempDir(), "missing-secrets")+`
pipelines:
  - id: reports
    source:
      backend: local
      path: `+sourceRoot+`
    destinations:
      - id: archive
        backend: local
        path: `+destinationRoot+`
`)

	err := Run(context.Background(), RunOptions{ConfigPath: configPath})
	if err == nil {
		t.Fatal("Run() error = nil, want secrets directory error")
	}
	if !strings.Contains(err.Error(), "load secrets directory") {
		t.Fatalf("Run() error = %v, want secrets directory error", err)
	}
	if strings.Contains(err.Error(), "missing-source") {
		t.Fatalf("Run() error = %v, opened source before loading secrets", err)
	}
}

func TestRunPrintsSecretConflictWarningWithoutValues(t *testing.T) {
	name := "DISTRIBUTOR_TEST_RUN_SECRET"
	t.Setenv(name, "process-value")
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	secretsRoot := t.TempDir()
	if err := os.WriteFile(filepath.Join(secretsRoot, name), []byte("secret-value\n"), 0o600); err != nil {
		t.Fatalf("write secret: %v", err)
	}
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	configPath := writeConfigFile(t, `
secrets:
  directory: `+secretsRoot+`
pipelines:
  - id: reports
    source:
      backend: local
      path: `+sourceRoot+`
    destinations:
      - id: archive
        backend: local
        path: `+destinationRoot+`
`)

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: configPath,
		DryRun:     true,
		Stdout:     &stdout,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	output := stdout.String()
	if !strings.Contains(output, "secret "+name+" ignored because the real environment already has that variable") {
		t.Fatalf("stdout = %q, want secret conflict warning", output)
	}
	if strings.Contains(output, "process-value") || strings.Contains(output, "secret-value") {
		t.Fatalf("stdout exposed secret values: %q", output)
	}
}

func TestWriteSSHWarningsReportsInsecureHostKeyPolicy(t *testing.T) {
	var stdout bytes.Buffer
	err := writeSSHWarnings(&stdout, config.Pipeline{
		ID: "reports",
		Source: config.Backend{
			Backend: config.BackendSSH,
			SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyOff},
		},
		Destinations: []config.Destination{{
			ID:      "archive",
			Backend: config.BackendSSH,
			SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyOff},
		}},
	})
	if err != nil {
		t.Fatalf("writeSSHWarnings() error = %v", err)
	}
	output := stdout.String()
	for _, want := range []string{
		"pipeline=reports source host_key_policy=off disables SSH host key checking",
		"pipeline=reports destination=archive host_key_policy=off disables SSH host key checking",
	} {
		if !strings.Contains(output, want) {
			t.Fatalf("output = %q, want substring %q", output, want)
		}
	}
}

func TestRunPublishesNewLocalBundle(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
		Stdout:     &stdout,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
	assertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
	if _, err := os.Stat(filepath.Join(destinationRoot, "manifest.json")); !os.IsNotExist(err) {
		t.Fatalf("destination manifest stat error = %v, want not exist", err)
	}
	destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
	if destinationState.PipelineID != "reports" || destinationState.DestinationID != "archive" {
		t.Fatalf("state identity = %s/%s", destinationState.PipelineID, destinationState.DestinationID)
	}
	if destinationState.Source.Manifest.ID != manifest.ID {
		t.Fatalf("state source id = %q, want %q", destinationState.Source.Manifest.ID, manifest.ID)
	}
	if got, want := len(destinationState.Outputs), 2; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
}

func TestRunNotifiesAfterPublication(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	notifier := &recordingNotifier{
		check: func() {
			if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil {
				t.Fatalf("state stat during notify: %v", err)
			}
		},
	}

	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
		Notifier:   notifier,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if got, want := len(notifier.events), 1; got != want {
		t.Fatalf("notification count = %d, want %d", got, want)
	}
	event := notifier.events[0]
	if event.PipelineID != "reports" || event.DestinationID != "archive" || event.BundleID == "" || event.Action != "publish_new" {
		t.Fatalf("notification event = %#v", event)
	}
	if got, want := len(event.Outputs), 2; got != want {
		t.Fatalf("notification output count = %d, want %d", got, want)
	}
}

func TestRunNotifiesAfterReplacement(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	older := manifest
	older.Created = older.Created.Add(-time.Hour)
	writeDestinationState(t, destinationRoot, "", older)
	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil {
		t.Fatalf("write old output: %v", err)
	}
	notifier := &recordingNotifier{}

	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
		Notifier:   notifier,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if got, want := len(notifier.events), 1; got != want {
		t.Fatalf("notification count = %d, want %d", got, want)
	}
	if notifier.events[0].Action != "replace_older" {
		t.Fatalf("notification action = %q, want replace_older", notifier.events[0].Action)
	}
}

func TestRunDoesNotNotifyForSkippedDestination(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	configPath := writeLocalConfig(t, sourceRoot, destinationRoot)
	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
		t.Fatalf("first Run() error = %v", err)
	}
	notifier := &recordingNotifier{}

	err := Run(context.Background(), RunOptions{ConfigPath: configPath, Notifier: notifier})
	if err != nil {
		t.Fatalf("second Run() error = %v", err)
	}
	if len(notifier.events) != 0 {
		t.Fatalf("notifications = %#v, want none", notifier.events)
	}
}

func TestRunDoesNotNotifyDuringDryRun(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	notifier := &recordingNotifier{}

	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
		DryRun:     true,
		Notifier:   notifier,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if len(notifier.events) != 0 {
		t.Fatalf("notifications = %#v, want none", notifier.events)
	}
	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
	}
}

func TestRunContinuesAfterDestinationFailure(t *testing.T) {
	sourceRoot := t.TempDir()
	firstDestination := t.TempDir()
	secondDestination := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	if err := os.WriteFile(filepath.Join(firstDestination, "unmanaged.txt"), []byte("data"), 0o600); err != nil {
		t.Fatalf("write unmanaged file: %v", err)
	}

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: writeFanoutConfig(t, sourceRoot, firstDestination, secondDestination),
		Stdout:     &stdout,
	})
	if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") {
		t.Fatalf("Run() error = %v, want unmanaged failure", err)
	}
	if !strings.Contains(err.Error(), "pipeline reports destination archive-one backend local bundle .") {
		t.Fatalf("Run() error = %v, want backend context", err)
	}
	output := stdout.String()
	for _, want := range []string{
		"destination=archive-one backend=local action=error",
		"destination=archive-two backend=local action=publish_new",
		"Final status: failed planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=1 dry_run=false",
	} {
		if !strings.Contains(output, want) {
			t.Fatalf("stdout = %q, want substring %q", output, want)
		}
	}
	assertFile(t, filepath.Join(secondDestination, "report.md"), "# Report\nSunny.\n")
}

func TestRunPublishesHTMLOnly(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithPolicy(t, sourceRoot, destinationRoot, false, true)})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFileContains(t, filepath.Join(destinationRoot, "report.html"), "<h1>Report</h1>")
	if _, err := os.Stat(filepath.Join(destinationRoot, "report.md")); !os.IsNotExist(err) {
		t.Fatalf("report.md stat error = %v, want not exist", err)
	}
	destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
	if got, want := len(destinationState.Outputs), 1; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
	output := destinationState.Outputs[0]
	if output.Kind != state.OutputKindGenerated || output.Transform != "markdown_to_html" || output.Path != "report.html" || output.SourcePath != "report.md" {
		t.Fatalf("generated output metadata = %#v", output)
	}
}

func TestRunPublishesHTMLIndexWithExplicitInput(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{
		ExtraFiles: []testFile{{Path: "notes.md", Data: "# Notes\nHidden.\n"}},
	})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, false, true, config.TransformModeIndex, "report.md")})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFileContains(t, filepath.Join(destinationRoot, "index.html"), "<h1>Report</h1>")
	if _, err := os.Stat(filepath.Join(destinationRoot, "report.html")); !os.IsNotExist(err) {
		t.Fatalf("report.html stat error = %v, want not exist", err)
	}
	destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
	if got, want := len(destinationState.Outputs), 1; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
	output := destinationState.Outputs[0]
	if output.Kind != state.OutputKindGenerated || output.Transform != "markdown_to_html" || output.Path != "index.html" || output.SourcePath != "report.md" {
		t.Fatalf("generated output metadata = %#v", output)
	}
}

func TestRunPublishesHTMLIndexWithSingleMarkdownFallback(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, false, true, config.TransformModeIndex, "")})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFileContains(t, filepath.Join(destinationRoot, "index.html"), "<h1>Report</h1>")
}

func TestRunFailsIndexModeWithAmbiguousMarkdownInput(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{
		ExtraFiles: []testFile{{Path: "notes.md", Data: "# Notes\n"}},
	})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, false, true, config.TransformModeIndex, "")})
	if err == nil || !strings.Contains(err.Error(), "multiple markdown source files") {
		t.Fatalf("Run() error = %v, want ambiguous input error", err)
	}
	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
	}
}

func TestRunPublishesSourceAndHTML(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithPolicy(t, sourceRoot, destinationRoot, true, true)})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
	assertFileContains(t, filepath.Join(destinationRoot, "report.html"), "<p>Sunny.</p>")
	assertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
	destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
	if got, want := len(destinationState.Outputs), 3; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
}

func TestRunDoesNotMutateSourceBundle(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	sourcePath := filepath.Join(sourceRoot, "report.md")
	before, err := os.ReadFile(sourcePath)
	if err != nil {
		t.Fatalf("read source before: %v", err)
	}

	err = Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithPolicy(t, sourceRoot, destinationRoot, true, true)})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	after, err := os.ReadFile(sourcePath)
	if err != nil {
		t.Fatalf("read source after: %v", err)
	}
	if string(after) != string(before) {
		t.Fatalf("source changed from %q to %q", before, after)
	}
}

func TestRunFailsOnOutputPathCollision(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{ExtraFiles: []testFile{{Path: "report.html", Data: "<p>source html</p>\n"}}})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithPolicy(t, sourceRoot, destinationRoot, true, true)})
	if err == nil || !strings.Contains(err.Error(), "destination output path collision") {
		t.Fatalf("Run() error = %v, want collision", err)
	}
	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
	}
}

func TestRunFailsOnIndexOutputPathCollision(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{
		ExtraFiles: []testFile{{Path: "index.html", Data: "<p>source index</p>\n"}},
	})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, true, true, config.TransformModeIndex, "report.md")})
	if err == nil || !strings.Contains(err.Error(), "destination output path collision") {
		t.Fatalf("Run() error = %v, want collision", err)
	}
	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
	}
}

func TestRunDryRunReportsGeneratedOutputs(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfigWithPolicy(t, sourceRoot, destinationRoot, false, true),
		DryRun:     true,
		Stdout:     &stdout,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if !strings.Contains(stdout.String(), "outputs=report.html") {
		t.Fatalf("stdout = %q, want generated output path", stdout.String())
	}
}

func TestRunDryRunReportsIndexOutputWithoutWriting(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, false, true, config.TransformModeIndex, ""),
		DryRun:     true,
		Stdout:     &stdout,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if !strings.Contains(stdout.String(), "outputs=index.html") {
		t.Fatalf("stdout = %q, want index output path", stdout.String())
	}
	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
	}
}

func TestRunSourceOnlyDoesNotWriteIndexOutput(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, true, false, config.TransformModeIndex, "")})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
	if _, err := os.Stat(filepath.Join(destinationRoot, "index.html")); !os.IsNotExist(err) {
		t.Fatalf("index.html stat error = %v, want not exist", err)
	}
}

func TestRunReplacesHTMLIndexOutput(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{
		Created: testutil.DefaultCreated,
		Files: []testFile{
			{Path: "report.md", Data: "# Report\nOld.\n"},
			{Path: "summary.txt", Data: "Summary\n"},
		},
	})
	configPath := writeLocalConfigWithMarkdownTransform(t, sourceRoot, destinationRoot, false, true, config.TransformModeIndex, "")
	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
		t.Fatalf("first Run() error = %v", err)
	}
	assertFileContains(t, filepath.Join(destinationRoot, "index.html"), "<p>Old.</p>")

	writeSourceBundle(t, sourceRoot, "", testBundleOptions{
		Created: testutil.DefaultCreated.Add(time.Hour),
		Files: []testFile{
			{Path: "report.md", Data: "# Report\nNew.\n"},
			{Path: "summary.txt", Data: "Summary\n"},
		},
	})

	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
		t.Fatalf("second Run() error = %v", err)
	}
	assertFileContains(t, filepath.Join(destinationRoot, "index.html"), "<p>New.</p>")
}

func TestRunSkipsWhenDestinationStateMatches(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	configPath := writeLocalConfig(t, sourceRoot, destinationRoot)
	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
		t.Fatalf("first Run() error = %v", err)
	}

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{ConfigPath: configPath, Stdout: &stdout})
	if err != nil {
		t.Fatalf("second Run() error = %v", err)
	}
	if !strings.Contains(stdout.String(), "action=skip_same") {
		t.Fatalf("stdout = %q, want skip_same", stdout.String())
	}
}

func TestRunReplacesOlderDestination(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	older := manifest
	older.Created = older.Created.Add(-time.Hour)
	writeDestinationState(t, destinationRoot, "", older)
	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil {
		t.Fatalf("write old output: %v", err)
	}

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), Stdout: &stdout})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if !strings.Contains(stdout.String(), "action=replace_older") {
		t.Fatalf("stdout = %q, want replace_older", stdout.String())
	}
	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
}

func TestRunSkipsNewerDestination(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	newer := manifest
	newer.Created = newer.Created.Add(time.Hour)
	writeDestinationState(t, destinationRoot, "", newer)
	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("newer\n"), 0o600); err != nil {
		t.Fatalf("write newer output: %v", err)
	}

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), Stdout: &stdout})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if !strings.Contains(stdout.String(), "action=skip_destination_newer") {
		t.Fatalf("stdout = %q, want skip_destination_newer", stdout.String())
	}
	assertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n")
}

func TestRunFailsOnConflict(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	manifest.ID = "other.source"
	writeDestinationState(t, destinationRoot, "", manifest)

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)})
	if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
		t.Fatalf("Run() error = %v, want fail_conflict", err)
	}
}

func TestRunFailsOnUnmanagedDestination(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("data"), 0o600); err != nil {
		t.Fatalf("write unmanaged file: %v", err)
	}

	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)})
	if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") {
		t.Fatalf("Run() error = %v, want fail_unmanaged", err)
	}
}

func TestRunForceReplacesUnmanagedDestination(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
	if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("old"), 0o600); err != nil {
		t.Fatalf("write unmanaged file: %v", err)
	}

	var stdout bytes.Buffer
	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
		Force:      true,
		Stdout:     &stdout,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if !strings.Contains(stdout.String(), "action=force_replace") {
		t.Fatalf("stdout = %q, want force_replace", stdout.String())
	}
	if _, err := os.Stat(filepath.Join(destinationRoot, "unmanaged.txt")); !os.IsNotExist(err) {
		t.Fatalf("unmanaged file stat error = %v, want not exist", err)
	}
	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
}

func TestRunFansOutToLocalDestinations(t *testing.T) {
	sourceRoot := t.TempDir()
	firstDestination := t.TempDir()
	secondDestination := t.TempDir()
	writeSourceBundle(t, sourceRoot, "daily", testBundleOptions{})

	err := Run(context.Background(), RunOptions{ConfigPath: writeFanoutConfig(t, sourceRoot, firstDestination, secondDestination)})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFile(t, filepath.Join(firstDestination, "daily", "report.md"), "# Report\nSunny.\n")
	assertFile(t, filepath.Join(secondDestination, "daily", "summary.txt"), "Summary\n")
}

func TestRunFansOutWithDifferentPublishPolicies(t *testing.T) {
	sourceRoot := t.TempDir()
	archiveDestination := t.TempDir()
	htmlDestination := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	err := Run(context.Background(), RunOptions{ConfigPath: writeMixedPolicyFanoutConfig(t, sourceRoot, archiveDestination, htmlDestination)})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFile(t, filepath.Join(archiveDestination, "report.md"), "# Report\nSunny.\n")
	assertFile(t, filepath.Join(archiveDestination, "summary.txt"), "Summary\n")
	if _, err := os.Stat(filepath.Join(archiveDestination, "report.html")); !os.IsNotExist(err) {
		t.Fatalf("archive report.html stat error = %v, want not exist", err)
	}
	assertFileContains(t, filepath.Join(htmlDestination, "report.html"), "<h1>Report</h1>")
	if _, err := os.Stat(filepath.Join(htmlDestination, "report.md")); !os.IsNotExist(err) {
		t.Fatalf("html report.md stat error = %v, want not exist", err)
	}
}

func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) {
	localSourceRoot := t.TempDir()
	writeSourceBundle(t, localSourceRoot, "", testBundleOptions{})
	s3Source := fake.New()
	testutil.WriteFakeSourceBundle(t, s3Source, "", testutil.BundleOptions{})
	sshSource := fake.New()
	testutil.WriteFakeSourceBundle(t, sshSource, "", testutil.BundleOptions{})
	s3Destination := fake.New()
	sshDestination := fake.New()
	s3ToLocalDestination := t.TempDir()
	sshToLocalDestination := t.TempDir()

	cfg := crossBackendConfig(localSourceRoot, s3ToLocalDestination, sshToLocalDestination)
	if err := config.Validate(cfg); err != nil {
		t.Fatalf("cross-backend config validation error = %v", err)
	}
	provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{
		"s3:source-bucket":      s3Source,
		"s3:destination-bucket": s3Destination,
		"ssh:/source":           sshSource,
		"ssh:/destination":      sshDestination,
	})

	var dryRunOutput bytes.Buffer
	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true, Stdout: &dryRunOutput}, provider); err != nil {
		t.Fatalf("dry-run error = %v", err)
	}
	for _, want := range []string{
		"pipeline=local-to-s3 source=local",
		"destination=object-archive backend=s3 action=publish_new",
		"pipeline=s3-to-local source=s3",
		"destination=local-archive backend=local action=publish_new",
		"pipeline=local-to-ssh source=local",
		"destination=ssh-archive backend=ssh action=publish_new",
		"pipeline=ssh-to-local source=ssh",
		"Final status: ok planned=4 publish_new=4 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
	} {
		if !strings.Contains(dryRunOutput.String(), want) {
			t.Fatalf("dry-run output = %q, want substring %q", dryRunOutput.String(), want)
		}
	}
	if hasAny, err := s3Destination.HasAny(context.Background(), ""); err != nil || hasAny {
		t.Fatalf("s3 destination after dry-run hasAny=%t err=%v, want empty", hasAny, err)
	}
	if entries, err := os.ReadDir(s3ToLocalDestination); err != nil || len(entries) != 0 {
		t.Fatalf("s3-to-local destination entries = %v err=%v, want empty", entries, err)
	}

	var publishOutput bytes.Buffer
	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Stdout: &publishOutput}, provider); err != nil {
		t.Fatalf("publish error = %v", err)
	}
	assertFakeFile(t, s3Destination, "report.md", "# Report\nSunny.\n")
	assertFakeFile(t, sshDestination, "summary.txt", "Summary\n")
	assertFile(t, filepath.Join(s3ToLocalDestination, "report.md"), "# Report\nSunny.\n")
	assertFile(t, filepath.Join(sshToLocalDestination, "summary.txt"), "Summary\n")

	var repeatOutput bytes.Buffer
	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Stdout: &repeatOutput}, provider); err != nil {
		t.Fatalf("repeat error = %v", err)
	}
	if got := strings.Count(repeatOutput.String(), "action=skip_same"); got != 4 {
		t.Fatalf("repeat output = %q, skip_same count = %d, want 4", repeatOutput.String(), got)
	}
}

func TestRunForceReplacementStaysWithinRemoteBundlePaths(t *testing.T) {
	localSourceRoot := t.TempDir()
	writeSourceBundle(t, localSourceRoot, "bundle", testBundleOptions{})
	s3Destination := fake.New()
	sshDestination := fake.New()
	mustWriteFake(t, s3Destination, "bundle/old.txt", "old")
	mustWriteFake(t, s3Destination, "bundle-sibling/keep.txt", "keep")
	mustWriteFake(t, sshDestination, "bundle/old.txt", "old")
	mustWriteFake(t, sshDestination, "bundle-sibling/keep.txt", "keep")
	cfg := config.Config{Pipelines: []config.Pipeline{{
		ID:     "reports",
		Source: config.Backend{Backend: config.BackendLocal, Path: localSourceRoot},
		Destinations: []config.Destination{
			{
				ID:       "object-archive",
				Backend:  config.BackendS3,
				Endpoint: "http://s3.test",
				Bucket:   "destination-bucket",
			},
			{
				ID:      "ssh-archive",
				Backend: config.BackendSSH,
				Host:    "ssh.test",
				Path:    "/destination",
			},
		},
	}}}
	config.ApplyDefaults(&cfg)
	provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{
		"s3:destination-bucket": s3Destination,
		"ssh:/destination":      sshDestination,
	})

	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Force: true}, provider); err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	assertFakeFile(t, s3Destination, "bundle/report.md", "# Report\nSunny.\n")
	assertFakeMissing(t, s3Destination, "bundle/old.txt")
	assertFakeFile(t, s3Destination, "bundle-sibling/keep.txt", "keep")
	assertFakeFile(t, sshDestination, "bundle/report.md", "# Report\nSunny.\n")
	assertFakeMissing(t, sshDestination, "bundle/old.txt")
	assertFakeFile(t, sshDestination, "bundle-sibling/keep.txt", "keep")
}

func TestRunDryRunDoesNotWrite(t *testing.T) {
	sourceRoot := t.TempDir()
	destinationRoot := t.TempDir()
	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})

	err := Run(context.Background(), RunOptions{
		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
		DryRun:     true,
	})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
	}
}

type testBundleOptions struct {
	ID         string
	Created    time.Time
	Files      []testFile
	ExtraFiles []testFile
}

type testFile struct {
	Path string
	Data string
}

func writeSourceBundle(t *testing.T, root, relative string, opts testBundleOptions) bundle.Manifest {
	t.Helper()
	var files []testutil.SourceFile
	if opts.Files != nil {
		files = make([]testutil.SourceFile, 0, len(opts.Files))
		for _, file := range opts.Files {
			files = append(files, testutil.SourceFile{Path: file.Path, Data: file.Data})
		}
	}
	extraFiles := make([]testutil.SourceFile, 0, len(opts.ExtraFiles))
	for _, file := range opts.ExtraFiles {
		extraFiles = append(extraFiles, testutil.SourceFile{Path: file.Path, Data: file.Data})
	}
	return testutil.WriteSourceBundle(t, root, relative, testutil.BundleOptions{
		ID:         opts.ID,
		Created:    opts.Created,
		Files:      files,
		ExtraFiles: extraFiles,
	})
}

func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
	t.Helper()
	return testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
}

func writeLocalConfigWithPolicy(t *testing.T, sourceRoot, destinationRoot string, publishSource, publishHTML bool) string {
	t.Helper()
	transformConfig := ""
	if publishHTML {
		transformConfig = `
        transform:
          markdown_to_html:
            enabled: true
            mode: sidecar`
	}
	return writeConfigFile(t, `
pipelines:
  - id: reports
    source:
      backend: local
      path: `+sourceRoot+`
    destinations:
      - id: archive
        backend: local
        path: `+destinationRoot+`
        publish:
          source: `+fmt.Sprintf("%t", publishSource)+`
          html: `+fmt.Sprintf("%t", publishHTML)+transformConfig+`
`)
}

func writeLocalConfigWithMarkdownTransform(t *testing.T, sourceRoot, destinationRoot string, publishSource, publishHTML bool, mode, input string) string {
	t.Helper()
	enabled := publishHTML
	inputConfig := ""
	if input != "" {
		inputConfig = `
            input: ` + input
	}
	return writeConfigFile(t, `
pipelines:
  - id: reports
    source:
      backend: local
      path: `+sourceRoot+`
    destinations:
      - id: archive
        backend: local
        path: `+destinationRoot+`
        publish:
          source: `+fmt.Sprintf("%t", publishSource)+`
          html: `+fmt.Sprintf("%t", publishHTML)+`
        transform:
          markdown_to_html:
            enabled: `+fmt.Sprintf("%t", enabled)+`
            mode: `+mode+inputConfig+`
`)
}

func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
	t.Helper()
	return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)
}

func writeMixedPolicyFanoutConfig(t *testing.T, sourceRoot, archiveDestination, htmlDestination string) string {
	t.Helper()
	return writeConfigFile(t, `
pipelines:
  - id: reports
    source:
      backend: local
      path: `+sourceRoot+`
    destinations:
      - id: archive
        backend: local
        path: `+archiveDestination+`
        publish:
          source: true
          html: false
      - id: html
        backend: local
        path: `+htmlDestination+`
        publish:
          source: false
          html: true
        transform:
          markdown_to_html:
            enabled: true
            mode: sidecar
`)
}

func writeConfigFile(t *testing.T, body string) string {
	t.Helper()
	path := filepath.Join(t.TempDir(), "config.yml")
	if err := os.WriteFile(path, []byte(strings.TrimSpace(body)+"\n"), 0o600); err != nil {
		t.Fatalf("write config: %v", err)
	}
	return path
}

func writeDestinationState(t *testing.T, root, relative string, manifest bundle.Manifest) {
	t.Helper()
	testutil.WriteDestinationState(t, root, relative, manifest, testutil.DestinationStateOptions{})
}

func readStateFile(t *testing.T, path string) state.DistributorState {
	t.Helper()
	return testutil.ReadDestinationState(t, path)
}

func assertFile(t *testing.T, path, want string) {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("read file %s: %v", path, err)
	}
	if got := string(data); got != want {
		t.Fatalf("%s = %q, want %q", path, got, want)
	}
}

func assertFileContains(t *testing.T, path, want string) {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("read file %s: %v", path, err)
	}
	if !strings.Contains(string(data), want) {
		t.Fatalf("%s = %q, want substring %q", path, data, want)
	}
}

func assertFakeFile(t *testing.T, backend *fake.Backend, path, want string) {
	t.Helper()
	data, err := backend.ReadFile(context.Background(), path)
	if err != nil {
		t.Fatalf("read fake file %s: %v", path, err)
	}
	if got := string(data); got != want {
		t.Fatalf("%s = %q, want %q", path, got, want)
	}
}

func assertFakeMissing(t *testing.T, backend *fake.Backend, path string) {
	t.Helper()
	if _, err := backend.Stat(context.Background(), path); !storage.IsNotFound(err) {
		t.Fatalf("fake file %s stat error = %v, want not found", path, err)
	}
}

func mustWriteFake(t *testing.T, backend *fake.Backend, path, data string) {
	t.Helper()
	if _, err := backend.WriteFile(context.Background(), path, []byte(data), storage.WriteOptions{}); err != nil {
		t.Fatalf("write fake file %s: %v", path, err)
	}
}

func crossBackendConfig(localSourceRoot, s3ToLocalDestination, sshToLocalDestination string) config.Config {
	cfg := config.Config{
		Pipelines: []config.Pipeline{
			{
				ID:     "local-to-s3",
				Source: config.Backend{Backend: config.BackendLocal, Path: localSourceRoot},
				Destinations: []config.Destination{{
					ID:       "object-archive",
					Backend:  config.BackendS3,
					Endpoint: "http://s3.test",
					Bucket:   "destination-bucket",
				}},
			},
			{
				ID: "s3-to-local",
				Source: config.Backend{
					Backend:  config.BackendS3,
					Endpoint: "http://s3.test",
					Bucket:   "source-bucket",
				},
				Destinations: []config.Destination{{
					ID:      "local-archive",
					Backend: config.BackendLocal,
					Path:    s3ToLocalDestination,
				}},
			},
			{
				ID:     "local-to-ssh",
				Source: config.Backend{Backend: config.BackendLocal, Path: localSourceRoot},
				Destinations: []config.Destination{{
					ID:      "ssh-archive",
					Backend: config.BackendSSH,
					Host:    "ssh.test",
					Path:    "/destination",
				}},
			},
			{
				ID: "ssh-to-local",
				Source: config.Backend{
					Backend: config.BackendSSH,
					Host:    "ssh.test",
					Path:    "/source",
				},
				Destinations: []config.Destination{{
					ID:      "local-archive",
					Backend: config.BackendLocal,
					Path:    sshToLocalDestination,
				}},
			},
		},
	}
	config.ApplyDefaults(&cfg)
	return cfg
}

func fakeBackendFactoryProvider(t *testing.T, remoteBackends map[string]storage.Backend) backendFactoryProvider {
	t.Helper()
	return func(environment config.Environment) *backendFactory {
		registry := storage.NewRegistry()
		if err := registry.Register(config.BackendLocal, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			return local.New(cfg[storagePathKey])
		}); err != nil {
			t.Fatalf("register local backend: %v", err)
		}
		if err := registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			backend := remoteBackends["s3:"+cfg[s3BucketKey]]
			if backend == nil {
				return nil, fmt.Errorf("missing fake s3 backend for bucket %s", cfg[s3BucketKey])
			}
			return backend, nil
		}); err != nil {
			t.Fatalf("register s3 backend: %v", err)
		}
		if err := registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			backend := remoteBackends["ssh:"+cfg[storagePathKey]]
			if backend == nil {
				return nil, fmt.Errorf("missing fake ssh backend for path %s", cfg[storagePathKey])
			}
			return backend, nil
		}); err != nil {
			t.Fatalf("register ssh backend: %v", err)
		}
		return &backendFactory{registry: registry, environment: environment}
	}
}

type recordingNotifier struct {
	events []notify.Event
	check  func()
}

func (n *recordingNotifier) Notify(ctx context.Context, event notify.Event) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if n.check != nil {
		n.check()
	}
	n.events = append(n.events, event)
	return nil
}