From 39e071f5caeb9f850e78d53244f7137d1f2e8d5e Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 5 Jul 2026 14:27:23 +0000 Subject: [PATCH] Add CLI reference binding flags --- docs/cli.md | 13 +- docs/config.md | 13 ++ internal/cli/run.go | 273 +++++++++++++++++++++++++++++++++++++- internal/cli/run_test.go | 278 ++++++++++++++++++++++++++++++++++++++- 4 files changed, 567 insertions(+), 10 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index fa92fe5..8f79706 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -20,7 +20,7 @@ a bearer token. ```text notarius help -notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] +notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference slot=path] [--without-reference slot] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ``` @@ -46,11 +46,22 @@ Flags: invocation. - `--llm-profile id`: override every effective module binding to use one LLM profile. +- `--reference slot=path`: bind a reference path to an extractor reference + slot. Repeatable. Use `lane.slot=path` when multiple selected lanes declare + the same slot. +- `--without-reference slot`: remove a configured optional reference binding. + Repeatable. Use `lane.slot` when multiple selected lanes declare the same + slot. On success, the command prints the completed pipeline ID, approved and rejected artifact counts, and the output directory. If the run completes with warnings, the warning count is printed to stderr. +Reference flags are resolved against selected artifact lanes before the run +starts. Flat slot names are accepted only when exactly one selected lane +declares that slot. Binding paths are recorded in resolved pipeline metadata; +reference file content is not read by the current runner. + For durable output, diagnostics, retention, and failure inspection, see [Operations](operations.md). diff --git a/docs/config.md b/docs/config.md index 0feb26a..4b947aa 100644 --- a/docs/config.md +++ b/docs/config.md @@ -121,6 +121,9 @@ Pipeline fields: - `artifacts`: required for pipeline resolution. It maps artifact lane IDs to lane definitions. - `output`: optional module binding. Default module is `json`. +- `references`: optional map of extractor reference slot names to reference + paths. These bindings are defaults for artifact lanes whose extractor declares + the matching slot. Artifact lane fields: @@ -129,11 +132,19 @@ Artifact lane fields: - `normalize`: optional module binding. Default module is `noop`. - `validators`: optional list of module bindings. The production CLI currently does not register validator modules. +- `references`: optional map of extractor reference slot names to reference + paths. Lane bindings override pipeline-level bindings for the same slot. `notarius run` and `notarius config validate --pipeline` resolve the pipeline against the production module catalog and fail fast for unknown or incompatible module keys. +Reference bindings are validated against extractor-declared slots during +pipeline resolution. Required slots must be bound after config defaults, +lane-level bindings, and run-time `--reference` or `--without-reference` +overrides are applied. Reference paths are recorded in resolved pipeline +metadata; the current runner does not read reference file content. + ## Module Bindings Every module binding may use shorthand: @@ -215,3 +226,5 @@ Pipeline resolution additionally checks: - required module keys are present; - module keys are registered for the expected slot; - module capability requirements are satisfied. +- bound reference slots are declared by selected lane extractors; +- required reference slots are bound for selected lanes. diff --git a/internal/cli/run.go b/internal/cli/run.go index f25ec8c..90703ff 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -25,7 +25,7 @@ const defaultOutputRoot = "./notarius-output" const usage = `Usage: notarius help - notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] + notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference slot=path] [--without-reference slot] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ` @@ -95,6 +95,10 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i outputDir := fs.String("output-dir", "", "output directory") diagnosticsDir := fs.String("diagnostics-dir", "", "diagnostics directory") llmProfile := fs.String("llm-profile", "", "LLM profile override") + referenceFlags := stringListFlag{} + withoutReferenceFlags := stringListFlag{} + fs.Var(&referenceFlags, "reference", "reference binding, as slot=path or lane.slot=path") + fs.Var(&withoutReferenceFlags, "without-reference", "unbind a reference, as slot or lane.slot") if err := fs.Parse(reorderRunArgs(args)); err != nil { fmt.Fprintf(stderr, "notarius: %v\n", err) return 2 @@ -121,6 +125,16 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i fmt.Fprintf(stderr, "notarius: %v\n", err) return 2 } + referenceRequests, err := parseReferenceFlags(referenceFlags) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 2 + } + referenceUnbindRequests, err := parseReferenceUnbindFlags(withoutReferenceFlags) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 2 + } cfg, loadedConfigPath, err := loadConfig(*configPath, opts) if err != nil { @@ -155,11 +169,17 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) } + referenceOverrides, referenceUnbinds, err := resolveCLIReferenceRequests(cfg, pipelineID, only, catalog, referenceRequests, referenceUnbindRequests) + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) + } effective, err := cfg.Resolve(config.ResolveInput{ PipelineID: pipelineID, Only: only, Catalog: catalog, LLMProfileOverride: *llmProfile, + ReferenceOverrides: referenceOverrides, + ReferenceUnbinds: referenceUnbinds, }) if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) @@ -412,7 +432,7 @@ func reorderRunArgs(args []string) []string { func runFlagTakesValue(arg string) bool { switch arg { - case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile": + case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile", "--reference", "--without-reference": return true default: return false @@ -661,6 +681,255 @@ func parseOnly(raw string) ([]string, error) { return result, nil } +type stringListFlag []string + +func (flag *stringListFlag) String() string { + if flag == nil { + return "" + } + return strings.Join(*flag, ",") +} + +func (flag *stringListFlag) Set(value string) error { + *flag = append(*flag, value) + return nil +} + +type cliReferenceRequest struct { + LaneID string + SlotName string + Source string +} + +type cliReferenceUnbindRequest struct { + LaneID string + SlotName string +} + +func parseReferenceFlags(values []string) ([]cliReferenceRequest, error) { + if len(values) == 0 { + return nil, nil + } + requests := make([]cliReferenceRequest, 0, len(values)) + for _, raw := range values { + name, source, ok := strings.Cut(raw, "=") + if !ok { + return nil, fmt.Errorf("--reference must use slot=path or lane.slot=path") + } + if strings.TrimSpace(source) == "" { + return nil, fmt.Errorf("--reference path must not be empty; use --without-reference to unbind") + } + laneID, slotName, err := parseReferenceSelector(name, "--reference") + if err != nil { + return nil, err + } + requests = append(requests, cliReferenceRequest{ + LaneID: laneID, + SlotName: slotName, + Source: strings.TrimSpace(source), + }) + } + return requests, nil +} + +func parseReferenceUnbindFlags(values []string) ([]cliReferenceUnbindRequest, error) { + if len(values) == 0 { + return nil, nil + } + requests := make([]cliReferenceUnbindRequest, 0, len(values)) + for _, raw := range values { + if strings.Contains(raw, "=") { + return nil, fmt.Errorf("--without-reference must use slot or lane.slot") + } + laneID, slotName, err := parseReferenceSelector(raw, "--without-reference") + if err != nil { + return nil, err + } + requests = append(requests, cliReferenceUnbindRequest{ + LaneID: laneID, + SlotName: slotName, + }) + } + return requests, nil +} + +func parseReferenceSelector(raw string, flagName string) (string, string, error) { + selector := strings.TrimSpace(raw) + if selector == "" { + return "", "", fmt.Errorf("%s reference slot must not be empty", flagName) + } + if strings.Count(selector, ".") > 1 { + return "", "", fmt.Errorf("%s must use slot or lane.slot", flagName) + } + laneID := "" + slotName := selector + if strings.Contains(selector, ".") { + before, after, _ := strings.Cut(selector, ".") + laneID = strings.TrimSpace(before) + slotName = strings.TrimSpace(after) + if laneID == "" || slotName == "" { + return "", "", fmt.Errorf("%s must use non-empty lane.slot values", flagName) + } + } + return laneID, slotName, nil +} + +func resolveCLIReferenceRequests( + cfg config.Config, + pipelineID string, + only []string, + catalog pipeline.ModuleCatalog, + referenceRequests []cliReferenceRequest, + unbindRequests []cliReferenceUnbindRequest, +) ([]pipeline.ReferenceBinding, []pipeline.ReferenceUnbind, error) { + if len(referenceRequests) == 0 && len(unbindRequests) == 0 { + return nil, nil, nil + } + + selected, err := selectedReferenceLanes(cfg, pipelineID, only, catalog) + if err != nil { + return nil, nil, err + } + + overrides := make([]pipeline.ReferenceBinding, 0, len(referenceRequests)) + for _, request := range referenceRequests { + laneID, err := resolveCLIReferenceLane(selected, request.LaneID, request.SlotName) + if err != nil { + return nil, nil, err + } + overrides = append(overrides, pipeline.ReferenceBinding{ + LaneID: laneID, + SlotName: request.SlotName, + Source: request.Source, + BindingSource: contracts.ReferenceBindingSourceCLI, + }) + } + + unbinds := make([]pipeline.ReferenceUnbind, 0, len(unbindRequests)) + for _, request := range unbindRequests { + laneID, err := resolveCLIReferenceLane(selected, request.LaneID, request.SlotName) + if err != nil { + return nil, nil, err + } + unbinds = append(unbinds, pipeline.ReferenceUnbind{ + LaneID: laneID, + SlotName: request.SlotName, + }) + } + + return overrides, unbinds, nil +} + +type selectedReferenceLane struct { + id string + slots map[string]struct{} +} + +func selectedReferenceLanes(cfg config.Config, pipelineID string, only []string, catalog pipeline.ModuleCatalog) ([]selectedReferenceLane, error) { + profile, ok := lookupCLIReferencePipeline(cfg.Pipelines, pipelineID) + if !ok { + return nil, fmt.Errorf("pipeline %q is not configured", strings.TrimSpace(pipelineID)) + } + + lanesByID := make(map[string]pipeline.ArtifactLaneProfile, len(profile.Artifacts)) + for rawLaneID, lane := range profile.Artifacts { + laneID := strings.TrimSpace(rawLaneID) + if laneID == "" { + return nil, fmt.Errorf("pipeline %q artifact lane id must not be empty", strings.TrimSpace(pipelineID)) + } + if _, ok := lanesByID[laneID]; ok { + return nil, fmt.Errorf("pipeline %q artifact lane %q is duplicated after trimming", strings.TrimSpace(pipelineID), laneID) + } + lanesByID[laneID] = lane + } + + selectedIDs := make([]string, 0, len(lanesByID)) + if len(only) == 0 { + for laneID := range lanesByID { + selectedIDs = append(selectedIDs, laneID) + } + } else { + seen := make(map[string]struct{}, len(only)) + for _, rawLaneID := range only { + laneID := strings.TrimSpace(rawLaneID) + if laneID == "" { + return nil, fmt.Errorf("pipeline %q selected artifact lane id must not be empty", strings.TrimSpace(pipelineID)) + } + if _, ok := lanesByID[laneID]; !ok { + return nil, fmt.Errorf("pipeline %q selected artifact lane %q is not declared", strings.TrimSpace(pipelineID), laneID) + } + if _, ok := seen[laneID]; !ok { + selectedIDs = append(selectedIDs, laneID) + seen[laneID] = struct{}{} + } + } + } + sort.Strings(selectedIDs) + + selected := make([]selectedReferenceLane, 0, len(selectedIDs)) + for _, laneID := range selectedIDs { + lane := lanesByID[laneID] + extractModule := strings.TrimSpace(lane.Extract.Module) + if extractModule == "" { + return nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", strings.TrimSpace(pipelineID), laneID) + } + if catalog.Extractors == nil { + return nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", strings.TrimSpace(pipelineID), laneID, extractModule, extractModule) + } + spec, ok := catalog.Extractors.Spec(extractModule) + if !ok { + return nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", strings.TrimSpace(pipelineID), laneID, extractModule, extractModule) + } + slotSet := make(map[string]struct{}, len(spec.ReferenceSlots)) + for _, slot := range spec.ReferenceSlots { + slotSet[slot.Name] = struct{}{} + } + selected = append(selected, selectedReferenceLane{id: laneID, slots: slotSet}) + } + return selected, nil +} + +func lookupCLIReferencePipeline(profiles map[string]pipeline.PipelineProfile, pipelineID string) (pipeline.PipelineProfile, bool) { + pipelineID = strings.TrimSpace(pipelineID) + for rawID, profile := range profiles { + if strings.TrimSpace(rawID) == pipelineID { + return profile, true + } + } + return pipeline.PipelineProfile{}, false +} + +func resolveCLIReferenceLane(selected []selectedReferenceLane, requestedLaneID string, slotName string) (string, error) { + slotName = strings.TrimSpace(slotName) + requestedLaneID = strings.TrimSpace(requestedLaneID) + if requestedLaneID != "" { + for _, lane := range selected { + if lane.id == requestedLaneID { + if _, ok := lane.slots[slotName]; !ok { + return "", fmt.Errorf("reference slot %q is not declared by selected lane %q", slotName, requestedLaneID) + } + return requestedLaneID, nil + } + } + return "", fmt.Errorf("reference lane %q is not selected", requestedLaneID) + } + + matches := make([]string, 0, 1) + for _, lane := range selected { + if _, ok := lane.slots[slotName]; ok { + matches = append(matches, lane.id) + } + } + switch len(matches) { + case 0: + return "", fmt.Errorf("reference slot %q is not declared by any selected lane", slotName) + case 1: + return matches[0], nil + default: + return "", fmt.Errorf("reference slot %q is declared by multiple selected lanes (%s); use lane.slot", slotName, strings.Join(matches, ", ")) + } +} + func sortedPipelineIDs(cfg config.Config) []string { ids := make([]string, 0, len(cfg.Pipelines)) for id := range cfg.Pipelines { diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index a59b032..d26a967 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "reflect" + "sort" "strings" "testing" @@ -737,6 +738,217 @@ func TestRunPipelineLLMProfileOverrideSelectsFactoryProfile(t *testing.T) { } } +func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--reference", "roster=./roster.yml", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), + }) + + if code != 1 || !strings.Contains(stderr.String(), "read input") { + t.Fatalf("RunWithOptions() code = %d stderr=%q, want read input failure after resolution", code, stderr.String()) + } + resolved := readResolvedPipeline(t, diagnosticsDir) + refs := resolved.ArtifactLanes[0].References + want := []pipeline.ReferenceBinding{ + {LaneID: "events", SlotName: "roster", Source: "./roster.yml", BindingSource: contracts.ReferenceBindingSourceCLI}, + } + if !reflect.DeepEqual(refs, want) { + t.Fatalf("resolved references = %#v, want %#v", refs, want) + } +} + +func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--only", "events,notes", + "--diagnostics-dir", diagnosticsDir, + "--reference", "notes.roster=./notes.yml", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), + }) + + if code != 1 || !strings.Contains(stderr.String(), "read input") { + t.Fatalf("RunWithOptions() code = %d stderr=%q, want read input failure after resolution", code, stderr.String()) + } + resolved := readResolvedPipeline(t, diagnosticsDir) + events := resolvedArtifactLane(t, resolved, "events") + if len(events.References) != 0 { + t.Fatalf("events references = %#v, want none", events.References) + } + notes := resolvedArtifactLane(t, resolved, "notes") + if len(notes.References) != 1 || notes.References[0].Source != "./notes.yml" { + t.Fatalf("notes references = %#v, want lane-qualified binding", notes.References) + } +} + +func TestRunPipelineReferenceFlagRejectsAmbiguousFlatSlot(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes")) + inputPath := writeSeriatimInput(t) + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--reference", "roster=./roster.yml", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "multiple selected lanes") || !strings.Contains(stderr.String(), "lane.slot") { + t.Fatalf("stderr = %q, want ambiguous reference error", stderr.String()) + } +} + +func TestRunPipelineReferenceFlagsRejectMalformedValues(t *testing.T) { + tests := []struct { + name string + args []string + want string + }{ + {name: "missing equals", args: []string{"--reference", "roster"}, want: "slot=path"}, + {name: "empty path", args: []string{"--reference", "roster="}, want: "path must not be empty"}, + {name: "empty slot", args: []string{"--reference", "=./roster.yml"}, want: "slot must not be empty"}, + {name: "too many selector parts", args: []string{"--reference", "a.b.c=./roster.yml"}, want: "lane.slot"}, + {name: "unbind with equals", args: []string{"--without-reference", "roster=./roster.yml"}, want: "slot or lane.slot"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + args := []string{"run", "example", "--config", configPath, "--input", inputPath} + args = append(args, test.args...) + + code := RunWithOptions(args, &stdout, &stderr, Options{Catalog: fakeCatalog(t)}) + if code != 2 { + t.Fatalf("RunWithOptions() code = %d, want 2", code) + } + if !strings.Contains(stderr.String(), test.want) { + t.Fatalf("stderr = %q, want substring %q", stderr.String(), test.want) + } + }) + } +} + +func TestRunPipelineWithoutReferenceRemovesOptionalConfigBinding(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAMLWithReferences("example", "events", map[string]string{"roster": "./config-roster.yml"})) + inputPath := filepath.Join(t.TempDir(), "missing.json") + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--without-reference", "roster", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), + }) + + if code != 1 || !strings.Contains(stderr.String(), "read input") { + t.Fatalf("RunWithOptions() code = %d stderr=%q, want read input failure after resolution", code, stderr.String()) + } + resolved := readResolvedPipeline(t, diagnosticsDir) + if refs := resolved.ArtifactLanes[0].References; len(refs) != 0 { + t.Fatalf("references = %#v, want unbound optional slot", refs) + } +} + +func TestRunPipelineWithoutReferenceFailsWhenRequiredSlotWouldBeMissing(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAMLWithReferences("example", "events", map[string]string{"roster": "./config-roster.yml"})) + inputPath := writeSeriatimInput(t) + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--without-reference", "roster", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster", Required: true}, + }, + }), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "required reference slot") || !strings.Contains(stderr.String(), "roster") { + t.Fatalf("stderr = %q, want required reference error", stderr.String()) + } +} + func TestRunPipelineWritesDurableOutputFiles(t *testing.T) { diagnosticsDir := t.TempDir() outputDir := t.TempDir() @@ -1305,6 +1517,27 @@ func testConfigYAMLForPipelines(pipelines map[string][]string) string { return b.String() } +func testConfigYAMLWithReferences(pipelineID string, laneID string, references map[string]string) string { + var b strings.Builder + b.WriteString("version: 1\n") + b.WriteString("pipelines:\n") + b.WriteString(" " + pipelineID + ":\n") + b.WriteString(" input: fake/input\n") + b.WriteString(" artifacts:\n") + b.WriteString(" " + laneID + ":\n") + b.WriteString(" extract: fake/extract\n") + b.WriteString(" references:\n") + keys := make([]string, 0, len(references)) + for key := range references { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + b.WriteString(" " + key + ": " + references[key] + "\n") + } + return b.String() +} + func mvpConfigYAML(pipelineID string, extractor string) string { return `version: 1 pipelines: @@ -1596,6 +1829,25 @@ func readJSONFile(t *testing.T, path string, out any) { } } +func readResolvedPipeline(t *testing.T, diagnosticsDir string) pipeline.ResolvedPipeline { + t.Helper() + runDir := onlyChildDir(t, diagnosticsDir) + var resolved pipeline.ResolvedPipeline + readJSONFile(t, filepath.Join(runDir, diagnostics.ArtifactResolvedPipeline), &resolved) + return resolved +} + +func resolvedArtifactLane(t *testing.T, resolved pipeline.ResolvedPipeline, laneID string) pipeline.ResolvedArtifactLane { + t.Helper() + for _, lane := range resolved.ArtifactLanes { + if lane.ID == laneID { + return lane + } + } + t.Fatalf("lane %q not found in resolved pipeline", laneID) + return pipeline.ResolvedArtifactLane{} +} + func assertNoTemporaryFiles(t *testing.T, root string) { t.Helper() if err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { @@ -1611,7 +1863,7 @@ func assertNoTemporaryFiles(t *testing.T, root string) { } } -func fakeCatalog(t *testing.T) pipeline.ModuleCatalog { +func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.ModuleCatalog { t.Helper() inputs := pipeline.NewInputAdapterRegistry() chunkers := pipeline.NewChunkerRegistry() @@ -1621,12 +1873,24 @@ func fakeCatalog(t *testing.T) pipeline.ModuleCatalog { validators := pipeline.NewValidatorRegistry() outputs := pipeline.NewOutputEncoderRegistry() - mustRegisterInput(t, inputs, pipeline.ModuleSpec{Key: "fake/input", Stage: pipeline.StageInput, Provides: []string{"source"}}) - mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{Key: "generic", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}}) - mustRegisterExtractor(t, extractors, pipeline.ModuleSpec{Key: "fake/extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}}) - mustRegisterMerger(t, mergers, pipeline.ModuleSpec{Key: "appendorder", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}}) - mustRegisterNormalizer(t, normalizers, pipeline.ModuleSpec{Key: "noop", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}}) - mustRegisterOutput(t, outputs, pipeline.ModuleSpec{Key: "json", Stage: pipeline.StageOutput, Requires: []string{"normalized"}}) + specs := map[string]pipeline.ModuleSpec{ + "fake/input": {Key: "fake/input", Stage: pipeline.StageInput, Provides: []string{"source"}}, + "generic": {Key: "generic", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}}, + "fake/extract": {Key: "fake/extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}}, + "appendorder": {Key: "appendorder", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}}, + "noop": {Key: "noop", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}}, + "json": {Key: "json", Stage: pipeline.StageOutput, Requires: []string{"normalized"}}, + } + for _, override := range overrides { + specs[override.Key] = override + } + + mustRegisterInput(t, inputs, specs["fake/input"]) + mustRegisterChunker(t, chunkers, specs["generic"]) + mustRegisterExtractor(t, extractors, specs["fake/extract"]) + mustRegisterMerger(t, mergers, specs["appendorder"]) + mustRegisterNormalizer(t, normalizers, specs["noop"]) + mustRegisterOutput(t, outputs, specs["json"]) return pipeline.ModuleCatalog{ Inputs: inputs,