eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Update run reporting for catalog workflows

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-19 15:50:09 +00:00
parent 52078e2195
commit 91b72478d5
9 changed files with 147 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
internal/app/run_test.go
View File

@@ -41,8 +41,8 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) {
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 replace_conflict=0 replace_newer=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true",
"bundle=. destination=archive backend=local action=publish_new workflow=additive outputs=report.md,summary.txt",
"Final status: ok planned=1 publish_new=1 upsert_additive=0 replace_catalog=0 skip_same=0 force_replace=0 fail_unmanaged=0 fail_conflict=0 failed=0 dry_run=true",
} {
if !strings.Contains(output, want) {
t.Fatalf("Run() output = %q, want substring %q", output, want)
@@ -496,7 +496,6 @@ func TestRunFixedPathDryRunReportsSelection(t *testing.T) {
}
func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
t.Skip("fixed-path replacement reporting is covered by the catalog reporting work")
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "old", testBundleOptions{
@@ -507,7 +506,7 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
{Path: "summary.txt", Data: "Old summary\n"},
},
})
configPath := testutil.WriteLocalConfigWithPathMapping(t, sourceRoot, destinationRoot, config.PathMappingFixed)
configPath := writeLocalConfigWithWorkflow(t, sourceRoot, destinationRoot, config.PathMappingFixed, config.WorkflowReplacement)
if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
t.Fatalf("first Run() error = %v", err)
}
@@ -531,9 +530,9 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
}
output := stdout.String()
for _, want := range []string{
"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_takeover takeover_mode=same_pipeline replaces destination root for selected_bundle=new reason=\"destination source id differs from source\"",
"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_takeover takeover_mode=same_pipeline outputs=report.md,summary.txt reason=\"destination source id differs from source\"",
"replace_takeover=1",
"Warning: pipeline=reports destination=archive path_mapping=fixed workflow=replacement action=replace_catalog replaces current-owner catalog outputs for selected_bundle=new",
"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_catalog workflow=replacement outputs=report.md,summary.txt reason=\"\"",
"replace_catalog=1",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -542,8 +541,7 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nOld.\n")
}
func TestRunJSONIncludesTakeoverActionAndSummary(t *testing.T) {
t.Skip("takeover reporting was replaced by catalog workflow reporting")
func TestRunJSONIncludesWorkflowActionAndSummary(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "old", testBundleOptions{
@@ -554,7 +552,7 @@ func TestRunJSONIncludesTakeoverActionAndSummary(t *testing.T) {
{Path: "summary.txt", Data: "Old summary\n"},
},
})
configPath := testutil.WriteLocalConfigWithPathMapping(t, sourceRoot, destinationRoot, config.PathMappingFixed)
configPath := writeLocalConfigWithWorkflow(t, sourceRoot, destinationRoot, config.PathMappingFixed, config.WorkflowReplacement)
if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
t.Fatalf("first Run() error = %v", err)
}
@@ -586,90 +584,18 @@ func TestRunJSONIncludesTakeoverActionAndSummary(t *testing.T) {
if !ok {
t.Fatalf("action = %#v, want object", actions[0])
}
if action["action"] != "replace_takeover" || action["takeover_mode"] != "same_pipeline" || action["reason"] != "destination source id differs from source" {
t.Fatalf("action = %#v, want takeover action metadata", action)
if action["action"] != "replace_catalog" || action["workflow"] != "replacement" || action["reason"] != nil {
t.Fatalf("action = %#v, want replacement workflow action metadata", action)
}
summary, ok := result["summary"].(map[string]any)
if !ok {
t.Fatalf("summary = %#v, want object", result["summary"])
}
if summary["replace_takeover"] != float64(1) || summary["replace_older"] != float64(0) || summary["force_replace"] != float64(0) {
t.Fatalf("summary = %#v, want takeover counter only", summary)
if summary["replace_catalog"] != float64(1) || summary["upsert_additive"] != float64(0) || summary["force_replace"] != float64(0) {
t.Fatalf("summary = %#v, want replacement workflow counter only", summary)
}
}
func TestRunFixedPathReplacesOlderManagedState(t *testing.T) {
t.Skip("legacy fixed-path replacement comparison no longer applies to catalog workflow")
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "old", testBundleOptions{
ID: "reports.old",
Created: testutil.DefaultCreated,
Files: []testFile{
{Path: "report.md", Data: "# Report\nOld.\n"},
{Path: "summary.txt", Data: "Old summary\n"},
},
})
configPath := testutil.WriteLocalConfigWithPathMapping(t, sourceRoot, destinationRoot, config.PathMappingFixed)
if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
t.Fatalf("first Run() error = %v", err)
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nOld.\n")
writeSourceBundle(t, sourceRoot, "new", testBundleOptions{
ID: "reports.new",
Created: testutil.DefaultCreated.Add(time.Hour),
Files: []testFile{
{Path: "report.md", Data: "# Report\nNew.\n"},
{Path: "summary.txt", Data: "New summary\n"},
},
})
var stdout bytes.Buffer
if err := Run(context.Background(), RunOptions{ConfigPath: configPath, Stdout: &stdout}); err != nil {
t.Fatalf("second Run() error = %v", err)
}
if !strings.Contains(stdout.String(), "action=replace_takeover takeover_mode=same_pipeline") {
t.Fatalf("stdout = %q, want same-pipeline takeover", stdout.String())
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nNew.\n")
destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
if destinationState.Source.Manifest.ID != "reports.new" {
t.Fatalf("state source id = %q, want reports.new", destinationState.Source.Manifest.ID)
}
}
func TestRunFixedPathSkipsWhenDestinationStateIsNewer(t *testing.T) {
t.Skip("legacy destination-newer comparison no longer applies to catalog workflow")
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
newer := testutil.ValidManifest(testutil.BundleOptions{
ID: "reports.same",
Created: testutil.DefaultCreated.Add(time.Hour),
})
writeDestinationState(t, destinationRoot, "", newer)
if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nExisting.\n"), 0o600); err != nil {
t.Fatalf("write existing report: %v", err)
}
writeSourceBundle(t, sourceRoot, "older", testBundleOptions{
ID: "reports.same",
Created: testutil.DefaultCreated,
})
var stdout bytes.Buffer
err := Run(context.Background(), RunOptions{
ConfigPath: testutil.WriteLocalConfigWithPathMapping(t, sourceRoot, destinationRoot, config.PathMappingFixed),
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())
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nExisting.\n")
}
func TestRunFixedPathFailsUnmanagedWithoutForce(t *testing.T) {
t.Skip("catalog workflow protects planned path collisions rather than unrelated unplanned content")
sourceRoot := t.TempDir()
@@ -846,20 +772,17 @@ func TestRunNotifiesGeneratedOutputMetadata(t *testing.T) {
}
func TestRunNotifiesAfterReplacement(t *testing.T) {
t.Skip("catalog workflow notification labels are covered by the catalog reporting work")
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)
writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
configPath := writeLocalConfigWithWorkflow(t, sourceRoot, destinationRoot, config.PathMappingPreserveRelative, config.WorkflowReplacement)
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: writeLocalConfig(t, sourceRoot, destinationRoot),
ConfigPath: configPath,
Notifier: notifier,
})
if err != nil {
@@ -868,8 +791,8 @@ func TestRunNotifiesAfterReplacement(t *testing.T) {
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)
if notifier.events[0].Action != "replace_catalog" {
t.Fatalf("notification action = %q, want replace_catalog", notifier.events[0].Action)
}
}
@@ -994,8 +917,8 @@ func TestBuildRunReportIncludesPartialFailures(t *testing.T) {
if got, want := len(report.Actions), 2; got != want {
t.Fatalf("action count = %d, want %d", got, want)
}
if report.Actions[0].DestinationID != "archive-one" || report.Actions[0].Action != "error" || !strings.Contains(report.Actions[0].Reason, "fail_unmanaged") {
t.Fatalf("first action = %#v, want archive-one error", report.Actions[0])
if report.Actions[0].DestinationID != "archive-one" || report.Actions[0].Action != "fail_unmanaged" || !strings.Contains(report.Actions[0].Reason, "fail_unmanaged") {
t.Fatalf("first action = %#v, want archive-one unmanaged failure", report.Actions[0])
}
if report.Actions[1].DestinationID != "archive-two" || report.Actions[1].Action != "publish_new" {
t.Fatalf("second action = %#v, want archive-two publish_new", report.Actions[1])
@@ -1130,7 +1053,7 @@ func TestRunStillRunsAllConfiguredPipelines(t *testing.T) {
testutil.AssertFile(t, filepath.Join(secondDestination, "report.md"), "# Report\nSunny.\n")
}
func TestRunDoesNotNotifyForSkippedDestination(t *testing.T) {
func TestRunNotifiesForAdditiveUpsert(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
@@ -1144,8 +1067,11 @@ func TestRunDoesNotNotifyForSkippedDestination(t *testing.T) {
if err != nil {
t.Fatalf("second Run() error = %v", err)
}
if len(notifier.events) != 0 {
t.Fatalf("notifications = %#v, want none", notifier.events)
if got, want := len(notifier.events), 1; got != want {
t.Fatalf("notification count = %d, want %d", got, want)
}
if notifier.events[0].Action != "upsert_additive" {
t.Fatalf("notification action = %q, want upsert_additive", notifier.events[0].Action)
}
}
@@ -1193,9 +1119,9 @@ func TestRunContinuesAfterDestinationFailure(t *testing.T) {
}
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 replace_conflict=0 replace_newer=0 replace_takeover=0 force_replace=0 skipped=0 failed=1 dry_run=false",
"destination=archive-one backend=local action=fail_unmanaged workflow=additive",
"destination=archive-two backend=local action=publish_new workflow=additive",
"Final status: failed planned=1 publish_new=1 upsert_additive=0 replace_catalog=0 skip_same=0 force_replace=0 fail_unmanaged=1 fail_conflict=0 failed=1 dry_run=false",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -1457,52 +1383,6 @@ func TestRunSkipsWhenDestinationStateMatches(t *testing.T) {
}
}
func TestRunReplacesOlderDestination(t *testing.T) {
t.Skip("legacy destination-older comparison no longer applies to catalog workflow")
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())
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
}
func TestRunSkipsNewerDestination(t *testing.T) {
t.Skip("legacy destination-newer comparison no longer applies to catalog workflow")
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())
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n")
}
func TestRunFailsOnUnmanagedDestination(t *testing.T) {
t.Skip("catalog workflow protects planned path collisions rather than unrelated unplanned content")
sourceRoot := t.TempDir()
@@ -1615,7 +1495,7 @@ func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) {
"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 replace_conflict=0 replace_newer=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true",
"Final status: ok planned=4 publish_new=4 upsert_additive=0 replace_catalog=0 skip_same=0 force_replace=0 fail_unmanaged=0 fail_conflict=0 failed=0 dry_run=true",
} {
if !strings.Contains(dryRunOutput.String(), want) {
t.Fatalf("dry-run output = %q, want substring %q", dryRunOutput.String(), want)
@@ -1746,6 +1626,24 @@ func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
return testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
}
func writeLocalConfigWithWorkflow(t *testing.T, sourceRoot, destinationRoot, pathMapping, workflow string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
workflow: `+workflow+`
path_mapping:
mode: `+pathMapping+`
`)
}
func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
t.Helper()
return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)