notarius/internal/core/config/pipeline_document_test.go

package config

import (
	"errors"
	"fmt"
	"testing"

	configkit "gitea.maximumdirect.net/eric/ConfigKit"
)

func TestDocumentRejectsExplicitInvalidBranches(t *testing.T) {
	t.Parallel()
	for _, slot := range []string{"input", "output"} {
		t.Run(slot+" empty validators", func(t *testing.T) {
			body := "input: {module: input, validators: []}"
			if slot == "output" {
				body = "input: {module: input}\n    output: {module: output, validators: []}"
			}
			_, err := loadConfigSource(configkit.Bytes("document.yml", []byte("version: 5\npipelines:\n  session:\n    "+body+"\n")), LoadOptions{})
			assertDocumentDiagnosticAt(t, err, "/pipelines/session/"+slot+"/validators")
		})
	}
	for _, value := range []string{
		"{path: ''}", "{path: '   '}",
		"{path: '', artifact: {step: previous, lane: registry}}",
		"{path: '   ', artifact: {step: previous, lane: registry}}",
		"{path: null, artifact: {step: previous, lane: registry}}",
	} {
		t.Run(value, func(t *testing.T) {
			content := fmt.Sprintf("version: 5\npipelines:\n  session:\n    input: {module: input}\n    artifacts:\n      lane:\n        extract:\n          module: extract\n          references:\n            party: %s\n", value)
			_, err := loadConfigSource(configkit.Bytes("document.yml", []byte(content)), LoadOptions{})
			path := "/pipelines/session/artifacts/lane/extract/references/party"
			if value == "{path: ''}" || value == "{path: '   '}" || value == "{path: null, artifact: {step: previous, lane: registry}}" {
				path += "/path"
			}
			assertDocumentDiagnosticAt(t, err, path)
		})
	}
}

func TestDocumentPreservesSupportedEmptyValidatorsAndReferenceForms(t *testing.T) {
	t.Parallel()
	loaded, err := loadConfigSource(configkit.Bytes("document.yml", []byte(`version: 5
pipelines:
  session:
    input: {module: input}
    chunk: {module: chunk, validators: []}
    artifacts:
      lane:
        validators: []
        extract:
          module: extract
          validators: []
          references:
            external: {path: /data/party.txt}
            generated: {artifact: {step: previous, lane: registry}}
        merge: {module: merge, validators: []}
        normalize: {module: normalize, validators: []}
`)), LoadOptions{})
	if err != nil {
		t.Fatal(err)
	}
	profile := loaded.Config.Pipelines["session"]
	lane := profile.Artifacts["lane"]
	for _, override := range []bool{profile.Chunk.Validators.Set, lane.Extract.Validators.Set, lane.Merge.Validators.Set, lane.Normalize.Validators.Set} {
		if !override {
			t.Fatal("supported empty validator override lost explicitness")
		}
	}
	if len(lane.Validators) != 0 || lane.Extract.References["external"].Path != "/data/party.txt" ||
		lane.Extract.References["generated"].Artifact.Lane != "registry" {
		t.Fatalf("lane = %#v", lane)
	}
}

func TestDocumentPipelineNormalizesCanonicalIdentifiers(t *testing.T) {
	t.Parallel()

	loaded, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(`version: 5
pipelines:
  " session ":
    llm_profile: " local "
    input:
      module: " seriatim "
    artifacts:
      " spells ":
        extract:
          module: " dnd/spells "
          structured_output_repair_attempts: 0
          validators: []
        merge:
          module: dnd/merge
          references:
            previous:
              artifact:
                step: " default "
                lane: " spells "
`)), LoadOptions{})
	if err != nil {
		var loadError *configkit.LoadError
		if errors.As(err, &loadError) {
			t.Fatalf("loadDocumentSource() diagnostics = %#v", loadError.Diagnostics())
		}
		t.Fatalf("loadDocumentSource() error = %v", err)
	}
	profile, ok := loaded.Config.Pipelines["session"]
	if !ok || profile.Input.Module != "seriatim" || profile.LLMProfile == nil || *profile.LLMProfile != "local" {
		t.Fatalf("normalized pipeline = %#v", loaded.Config.Pipelines)
	}
	lane, ok := profile.Artifacts["spells"]
	if !ok || lane.Extract.StructuredOutputRepairAttempts == nil || *lane.Extract.StructuredOutputRepairAttempts != 0 || lane.Extract.Validators == nil || len(lane.Extract.Validators) != 0 {
		t.Fatalf("normalized lane = %#v", profile.Artifacts)
	}
	selector := lane.Merge.References["previous"].Artifact
	if selector == nil || selector.Step != "default" || selector.Lane != "spells" {
		t.Fatalf("artifact selector = %#v", selector)
	}
}

func TestDocumentPipelineRejectsInvalidShapes(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name   string
		source string
		path   string
	}{
		{
			name: "both pipeline forms",
			source: `version: 5
pipelines:
  session:
    input: {module: seriatim}
    artifacts: {}
    steps:
      - id: first
        artifacts: {}
`,
			path: "/pipelines/session/steps",
		},
		{
			name: "empty ordered steps",
			source: `version: 5
pipelines:
  session:
    input: {module: seriatim}
    steps: []
`,
			path: "/pipelines/session/steps",
		},
		{
			name: "normalized pipeline collision",
			source: `version: 5
pipelines:
  session: {input: {module: seriatim}}
  " session ": {input: {module: seriatim}}
`,
			path: "/pipelines/session",
		},
		{
			name: "duplicate lane across ordered steps",
			source: `version: 5
pipelines:
  session:
    input: {module: seriatim}
    steps:
      - id: first
        artifacts: {summary: {extract: {module: dnd/summary}}}
      - id: second
        artifacts: {summary: {extract: {module: dnd/summary}}}
`,
			path: "/pipelines/session/steps/1/artifacts/summary",
		},
		{
			name: "invalid reference union",
			source: `version: 5
pipelines:
  session:
    input: {module: seriatim}
    references:
      party: {path: ./party.txt, artifact: {step: one, lane: summary}}
`,
			path: "/pipelines/session/references/party",
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			_, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(test.source)), LoadOptions{})
			assertDocumentDiagnosticAt(t, err, test.path)
		})
	}
}

func TestDocumentPipelineRejectsInvalidBindingPlacementAndPresence(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name   string
		source string
		path   string
	}{
		{
			name: "input validators",
			source: `version: 5
pipelines:
  session:
    input:
      module: seriatim
      validators: [{module: generic/valid_json}]
`,
			path: "/pipelines/session/input/validators",
		},
		{
			name: "blank selected profile",
			source: `version: 5
pipelines:
  session:
    llm_profile: "   "
    input: {module: seriatim}
`,
			path: "/pipelines/session/llm_profile",
		},
		{
			name: "null validators",
			source: `version: 5
pipelines:
  session:
    input: {module: seriatim}
    chunk: {module: generic/chunk, validators: null}
`,
			path: "/pipelines/session/chunk/validators",
		},
		{
			name: "negative retry budget",
			source: `version: 5
pipelines:
  session:
    input: {module: seriatim, retries: -1}
`,
			path: "/pipelines/session/input/retries",
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			_, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(test.source)), LoadOptions{})
			assertDocumentDiagnosticAt(t, err, test.path)
		})
	}
}

func assertDocumentDiagnosticAt(t *testing.T, err error, path string) {
	t.Helper()
	var loadError *configkit.LoadError
	if !errors.As(err, &loadError) {
		t.Fatalf("error = %v, want ConfigKit load error", err)
	}
	for _, diagnostic := range loadError.Diagnostics() {
		if diagnostic.Path.String() == path {
			return
		}
	}
	t.Fatalf("diagnostics = %#v, want a diagnostic at %s", loadError.Diagnostics(), path)
}