distributor/internal/app/run_test.go

package app

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

	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
	"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/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 action=publish_new outputs=report.md,summary.txt",
		"Final status: ok planned=1 publish_new=1 replace_older=0 skipped=0 failed=0 dry_run=true",
	} {
		if !strings.Contains(output, want) {
			t.Fatalf("Run() 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)
	}
	output := stdout.String()
	for _, want := range []string{
		"destination=archive-one action=error",
		"destination=archive-two action=publish_new",
		"Final status: failed planned=1 publish_new=1 replace_older=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 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 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 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 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 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
	ExtraFiles []testFile
}

type testFile struct {
	Path string
	Data string
}

func writeSourceBundle(t *testing.T, root, relative string, opts testBundleOptions) bundle.Manifest {
	t.Helper()
	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,
		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 writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
	t.Helper()
	return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)
}

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

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
}