Add configuration validation and resolution contracts

This commit is contained in:
2026-07-18 15:31:05 +00:00
parent 86ebb62f84
commit d3a8dc7930
5 changed files with 1224 additions and 0 deletions

View File

@@ -0,0 +1,514 @@
package config
import (
"context"
"encoding/json"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestEffectiveConfigRejectsEmptyAndUnknownPipelineIDs(t *testing.T) {
cfg := configForEffectiveTests(t, effectiveProfile())
for _, pipelineID := range []string{"", "missing"} {
name := pipelineID
if name == "" {
name = "empty"
}
t.Run(name, func(t *testing.T) {
_, err := cfg.Resolve(ResolveInput{PipelineID: pipelineID, Catalog: effectiveCatalog(t)})
if err == nil || !strings.Contains(err.Error(), "pipeline") {
t.Fatalf("Resolve(%q) error = %v, want pipeline context", pipelineID, err)
}
})
}
}
func TestEffectiveConfigResolvesTrimmedPipelineMapKeys(t *testing.T) {
profile := effectiveProfile()
profile.ID = " main "
cfg := Default()
cfg.Pipelines = map[string]pipeline.PipelineProfile{" main ": profile}
effective, err := cfg.Resolve(ResolveInput{PipelineID: "main", Catalog: effectiveCatalog(t)})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
if effective.PipelineID != "main" || effective.ResolvedPipeline.ID != "main" {
t.Fatalf("resolved IDs = %q, %q", effective.PipelineID, effective.ResolvedPipeline.ID)
}
}
func TestEffectiveConfigOnlySelectsRequestedLanesWithoutMutatingSource(t *testing.T) {
profile := effectiveProfile()
profile.Artifacts["other"] = pipeline.ArtifactLaneProfile{Extract: pipeline.Binding("extract")}
cfg := configForEffectiveTests(t, profile)
effective, err := cfg.Resolve(ResolveInput{
PipelineID: "main",
Only: []string{"other"},
Catalog: effectiveCatalog(t),
})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
if len(effective.ResolvedPipeline.ArtifactLanes) != 1 || effective.ResolvedPipeline.ArtifactLanes[0].ID != "other" {
t.Fatalf("resolved lanes = %#v", effective.ResolvedPipeline.ArtifactLanes)
}
if len(cfg.Pipelines["main"].Artifacts) != 2 {
t.Fatalf("source lanes were mutated: %#v", cfg.Pipelines["main"].Artifacts)
}
_, err = cfg.Resolve(ResolveInput{
PipelineID: "main",
Only: []string{"missing"},
Catalog: effectiveCatalog(t),
})
if err == nil || !strings.Contains(err.Error(), "lane \"missing\"") {
t.Fatalf("unknown lane error = %v", err)
}
}
func TestEffectiveConfigMaterializesDefaultBindingsThroughCatalog(t *testing.T) {
effective, err := resolveEffectiveProfile(t, effectiveProfile(), ResolveInput{})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
resolved := effective.ResolvedPipeline
if resolved.Chunk.Module != pipeline.DefaultChunkModule || resolved.Output.Module != pipeline.DefaultOutputModule {
t.Fatalf("default pipeline bindings = %#v, %#v", resolved.Chunk, resolved.Output)
}
if len(resolved.ArtifactLanes) != 1 || resolved.ArtifactLanes[0].Merge.Module != pipeline.DefaultMergeModule || resolved.ArtifactLanes[0].Normalize.Module != pipeline.DefaultNormalizeModule {
t.Fatalf("default lane bindings = %#v", resolved.ArtifactLanes)
}
}
func TestEffectiveConfigResolutionFailuresRetainContext(t *testing.T) {
tests := []struct {
name string
mutate func(*pipeline.PipelineProfile)
want []string
}{
{
name: "unknown module",
mutate: func(profile *pipeline.PipelineProfile) {
profile.Input.Module = "missing-input"
},
want: []string{"pipeline \"main\"", "input"},
},
{
name: "missing capability",
mutate: func(profile *pipeline.PipelineProfile) {
profile.Chunk.Module = "needs-capability"
},
want: []string{"pipeline \"main\"", "chunk"},
},
{
name: "missing artifact variant",
mutate: func(profile *pipeline.PipelineProfile) {
profile.Artifacts["lane"] = pipeline.ArtifactLaneProfile{
Extract: pipeline.Binding("extract"),
Merge: pipeline.Binding("other-merge"),
}
},
want: []string{"pipeline \"main\"", "lane \"lane\"", "merge"},
},
{
name: "invalid module options",
mutate: func(profile *pipeline.PipelineProfile) {
profile.Chunk = pipeline.ModuleBinding{Module: "generic", Options: map[string]any{"unknown": true}}
},
want: []string{"pipeline \"main\"", "chunk", "generic", "options"},
},
{
name: "invalid validator options",
mutate: func(profile *pipeline.PipelineProfile) {
lane := profile.Artifacts["lane"]
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "option-validator",
Options: map[string]any{"invalid": true},
}},
}
profile.Artifacts["lane"] = lane
},
want: []string{"pipeline \"main\"", "lane \"lane\"", "extract", "option-validator", "options"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
profile := effectiveProfile()
tt.mutate(&profile)
_, err := resolveEffectiveProfile(t, profile, ResolveInput{})
if err == nil {
t.Fatal("Resolve() error = nil, want failure")
}
for _, fragment := range tt.want {
if !strings.Contains(err.Error(), fragment) {
t.Fatalf("Resolve() error = %v, want context %q", err, fragment)
}
}
})
}
}
func TestEffectiveConfigLLMProfileOverrideChangesDigestWithoutOverridingValidators(t *testing.T) {
profile := effectiveProfile()
profile.Chunk.LLMProfile = "chunk-profile"
lane := profile.Artifacts["lane"]
lane.Extract.LLMProfile = "extract-profile"
lane.Merge.LLMProfile = "merge-profile"
lane.Normalize.LLMProfile = "normalize-profile"
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "llm-validator",
LLMProfile: "validator-profile",
}},
}
profile.Artifacts["lane"] = lane
base, err := resolveEffectiveProfile(t, profile, ResolveInput{})
if err != nil {
t.Fatalf("base Resolve() error = %v", err)
}
overridden, err := resolveEffectiveProfile(t, profile, ResolveInput{LLMProfileOverride: "override-profile"})
if err != nil {
t.Fatalf("overridden Resolve() error = %v", err)
}
if base.ResolvedPipeline.Digest == overridden.ResolvedPipeline.Digest {
t.Fatal("LLM profile override did not change the pipeline digest")
}
resolved := overridden.ResolvedPipeline
if resolved.Chunk.LLMProfile != "override-profile" || resolved.ArtifactLanes[0].Extract.LLMProfile != "override-profile" ||
resolved.ArtifactLanes[0].Merge.LLMProfile != "override-profile" || resolved.ArtifactLanes[0].Normalize.LLMProfile != "override-profile" {
t.Fatalf("pipeline profile override was not applied: %#v", resolved)
}
validators := findEffectiveValidatorChain(resolved, pipeline.StageExtract, "lane")
if len(validators.Validators) != 1 || validators.Validators[0].Binding.LLMProfile != "validator-profile" {
t.Fatalf("validator profile was overridden: %#v", validators)
}
}
func TestEffectiveConfigValidatorOverridesRemainDistinctAndOrdered(t *testing.T) {
tests := []struct {
name string
value pipeline.ValidatorOverride
want []string
}{
{
name: "omitted uses default",
want: []string{"default-validator"},
},
{
name: "explicit empty",
value: pipeline.ValidatorOverride{Set: true},
want: nil,
},
{
name: "configured order",
value: pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{
pipeline.Binding("configured-a"),
pipeline.Binding("configured-b"),
},
},
want: []string{"configured-a", "configured-b"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
profile := effectiveProfile()
lane := profile.Artifacts["lane"]
lane.Extract.Validators = tt.value
profile.Artifacts["lane"] = lane
effective, err := resolveEffectiveProfile(t, profile, ResolveInput{})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
chain := findEffectiveValidatorChain(effective.ResolvedPipeline, pipeline.StageExtract, "lane")
got := make([]string, len(chain.Validators))
for i, validator := range chain.Validators {
got[i] = validator.Binding.Module
}
if len(got) != len(tt.want) {
t.Fatalf("validator chain = %#v, want %v", got, tt.want)
}
for i := range got {
if got[i] != tt.want[i] {
t.Fatalf("validator chain = %#v, want %v", got, tt.want)
}
}
})
}
}
func TestEffectiveConfigAndResolutionInputsDoNotAliasSource(t *testing.T) {
profile := effectiveProfile()
profile.Chunk.Options = map[string]any{"nested": map[string]any{"safe": "source"}}
profile.Chunk.References = map[string]string{"chunk-ref": "chunk.txt"}
lane := profile.Artifacts["lane"]
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "configured-a",
Options: map[string]any{"nested": map[string]any{"safe": "validator-source"}},
}},
}
profile.Artifacts["lane"] = lane
cfg := configForEffectiveTests(t, profile)
only := []string{"lane"}
overrides := []pipeline.ReferenceBinding{{Stage: pipeline.StageChunk, SlotName: "chunk-ref", Source: "source.txt"}}
effective, err := cfg.Resolve(ResolveInput{
PipelineID: "main",
Only: only,
ReferenceOverrides: overrides,
Catalog: effectiveCatalog(t),
})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
effective.Config.Pipelines["main"].Chunk.Options["nested"].(map[string]any)["safe"] = "effective-config"
effective.ResolvedPipeline.Chunk.Options["nested"].(map[string]any)["safe"] = "resolved-pipeline"
effective.ResolvedPipeline.ChunkReferences.Bindings[0].Source = "resolved-reference"
effective.ResolvedPipeline.ValidatorChains[1].Validators[0].Binding.Options["nested"].(map[string]any)["safe"] = "resolved-validator"
effective.Only[0] = "mutated-only"
effective.ReferenceOverrides[0].Source = "mutated-override"
if got := cfg.Pipelines["main"].Chunk.Options["nested"].(map[string]any)["safe"]; got != "source" {
t.Fatalf("source config option was aliased: %v", got)
}
if got := cfg.Pipelines["main"].Chunk.References["chunk-ref"]; got != "chunk.txt" {
t.Fatalf("source config references were aliased: %v", got)
}
if only[0] != "lane" || overrides[0].Source != "source.txt" {
t.Fatal("resolution inputs were aliased")
}
}
func effectiveProfile() pipeline.PipelineProfile {
return pipeline.PipelineProfile{
ID: "main",
Input: pipeline.Binding("input"),
Artifacts: map[string]pipeline.ArtifactLaneProfile{
"lane": {Extract: pipeline.Binding("extract")},
},
}
}
func configForEffectiveTests(t *testing.T, profile pipeline.PipelineProfile) Config {
t.Helper()
cfg := Default()
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
return cfg
}
func resolveEffectiveProfile(t *testing.T, profile pipeline.PipelineProfile, input ResolveInput) (EffectiveConfig, error) {
t.Helper()
cfg := configForEffectiveTests(t, profile)
if input.PipelineID == "" {
input.PipelineID = "main"
}
if input.Catalog.Inputs == nil {
input.Catalog = effectiveCatalog(t)
}
return cfg.Resolve(input)
}
func findEffectiveValidatorChain(resolved pipeline.ResolvedPipeline, stage pipeline.ModuleStage, laneID string) pipeline.ResolvedValidatorChain {
for _, chain := range resolved.ValidatorChains {
if chain.Stage == stage && chain.LaneID == laneID {
return chain
}
}
return pipeline.ResolvedValidatorChain{}
}
type effectiveArtifact struct {
Value string `json:"value"`
}
const effectiveArtifactKind contracts.ArtifactKind = "test/effective"
type effectiveCodec struct{}
func (effectiveCodec) Kind() contracts.ArtifactKind { return effectiveArtifactKind }
func (effectiveCodec) Schema() contracts.ArtifactSchema {
return contracts.ArtifactSchema{
ID: "effective-schema",
Name: "Effective artifact",
Version: "1",
JSONSchema: []byte(`{"type":"object"}`),
}
}
func (effectiveCodec) MediaType() string { return "application/json" }
func (effectiveCodec) EncodeCandidate(value effectiveArtifact) ([]byte, error) {
return json.Marshal(value)
}
func (effectiveCodec) Encode(value effectiveArtifact) ([]byte, error) {
return json.Marshal(value)
}
func (effectiveCodec) Decode(content []byte) (effectiveArtifact, error) {
var value effectiveArtifact
err := json.Unmarshal(content, &value)
return value, err
}
type effectiveInput struct{ key string }
func (m effectiveInput) Key() string { return m.key }
func (m effectiveInput) Parse(context.Context, contracts.ParseRequest) (*source.SourceDocument, error) {
return &source.SourceDocument{}, nil
}
type effectiveChunker struct{ key string }
func (m effectiveChunker) Key() string { return m.key }
func (m effectiveChunker) ReferenceSlots() []contracts.ReferenceSlot {
if m.key == pipeline.DefaultChunkModule {
return []contracts.ReferenceSlot{{Name: "chunk-ref"}}
}
return nil
}
func (m effectiveChunker) Plan(context.Context, contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
return contracts.ChunkPlanResult{}, nil
}
type effectiveExtractor struct{ key string }
func (m effectiveExtractor) Key() string { return m.key }
func (m effectiveExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil }
func (m effectiveExtractor) Extract(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[effectiveArtifact], error) {
return contracts.TypedExtractionResult[effectiveArtifact]{}, nil
}
type effectiveMerger struct{ key string }
func (m effectiveMerger) Key() string { return m.key }
func (m effectiveMerger) Merge(context.Context, contracts.TypedMergeRequest[effectiveArtifact]) (contracts.TypedMergeResult[effectiveArtifact], error) {
return contracts.TypedMergeResult[effectiveArtifact]{}, nil
}
type effectiveNormalizer struct{ key string }
func (m effectiveNormalizer) Key() string { return m.key }
func (m effectiveNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
func (m effectiveNormalizer) Normalize(context.Context, contracts.TypedNormalizeRequest[effectiveArtifact]) (contracts.TypedNormalizeResult[effectiveArtifact], error) {
return contracts.TypedNormalizeResult[effectiveArtifact]{}, nil
}
type effectiveOutput struct{ key string }
func (m effectiveOutput) Key() string { return m.key }
func (m effectiveOutput) Encode(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) {
return contracts.OutputResult{}, nil
}
type effectiveValidator struct {
name string
class contracts.ExecutionClass
}
func (v effectiveValidator) Name() string { return v.name }
func (v effectiveValidator) ExecutionClass() contracts.ExecutionClass { return v.class }
func (v effectiveValidator) Validate(context.Context, contracts.TypedValidationRequest[effectiveArtifact]) (contracts.ValidationResult, error) {
return contracts.ValidationResult{Approved: true}, nil
}
func effectiveCatalog(t *testing.T) pipeline.ModuleCatalog {
t.Helper()
catalog := pipeline.ModuleCatalog{
Inputs: pipeline.NewInputAdapterRegistry(),
Chunkers: pipeline.NewChunkerRegistry(),
ArtifactCodecs: pipeline.NewArtifactCodecRegistry(),
Extractors: pipeline.NewExtractorRegistry(),
Mergers: pipeline.NewMergerRegistry(),
Normalizers: pipeline.NewNormalizerRegistry(),
Validators: pipeline.NewValidatorRegistry(),
ValidatorChains: pipeline.NewValidatorChainRegistry(),
Outputs: pipeline.NewOutputEncoderRegistry(),
}
if err := pipeline.RegisterArtifactCodec(catalog.ArtifactCodecs, effectiveCodec{}); err != nil {
t.Fatal(err)
}
if err := catalog.Inputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "input", Stage: pipeline.StageInput, Provides: []string{"source"}}, func() (contracts.InputAdapter, error) {
return effectiveInput{key: "input"}, nil
}); err != nil {
t.Fatal(err)
}
chunkSpec := pipeline.ModuleSpec{
Key: pipeline.DefaultChunkModule,
Stage: pipeline.StageChunk,
Requires: []string{"source"},
Provides: []string{"chunk"},
ReferenceSlots: []contracts.ReferenceSlot{{Name: "chunk-ref"}},
}
chunkOptions := func(options map[string]any) error { return pipeline.RejectUnknownOptions(options, "size", "nested") }
if err := catalog.Chunkers.RegisterBuilderWithSpec(chunkSpec, chunkOptions, func(pipeline.BuildRequest) (contracts.Chunker, error) {
return effectiveChunker{key: pipeline.DefaultChunkModule}, nil
}); err != nil {
t.Fatal(err)
}
if err := catalog.Chunkers.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "needs-capability", Stage: pipeline.StageChunk, Requires: []string{"missing"}}, chunkOptions, func(pipeline.BuildRequest) (contracts.Chunker, error) {
return effectiveChunker{key: "needs-capability"}, nil
}); err != nil {
t.Fatal(err)
}
if err := pipeline.RegisterExtractor(catalog.Extractors, pipeline.ModuleSpec{Key: "extract", Stage: pipeline.StageExtract, ArtifactKind: effectiveArtifactKind, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, func() (contracts.Extractor[effectiveArtifact], error) {
return effectiveExtractor{key: "extract"}, nil
}); err != nil {
t.Fatal(err)
}
if err := pipeline.RegisterMerger(catalog.Mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: effectiveArtifactKind, Requires: []string{"candidate"}, Provides: []string{"merged"}}, func() (contracts.Merger[effectiveArtifact], error) {
return effectiveMerger{key: pipeline.DefaultMergeModule}, nil
}); err != nil {
t.Fatal(err)
}
if err := pipeline.RegisterMerger(catalog.Mergers, pipeline.ModuleSpec{Key: "other-merge", Stage: pipeline.StageMerge, ArtifactKind: "other-kind", Requires: []string{"candidate"}, Provides: []string{"merged"}}, func() (contracts.Merger[effectiveArtifact], error) {
return effectiveMerger{key: "other-merge"}, nil
}); err != nil {
t.Fatal(err)
}
if err := pipeline.RegisterNormalizer(catalog.Normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: effectiveArtifactKind, Requires: []string{"merged"}, Provides: []string{"normalized"}}, func() (contracts.Normalizer[effectiveArtifact], error) {
return effectiveNormalizer{key: pipeline.DefaultNormalizeModule}, nil
}); err != nil {
t.Fatal(err)
}
if err := catalog.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput, Requires: []string{"normalized"}}, func() (contracts.OutputEncoder, error) {
return effectiveOutput{key: pipeline.DefaultOutputModule}, nil
}); err != nil {
t.Fatal(err)
}
for _, validator := range []struct {
key string
class contracts.ExecutionClass
}{
{key: "default-validator", class: contracts.ExecutionClassDeterministic},
{key: "configured-a", class: contracts.ExecutionClassDeterministic},
{key: "configured-b", class: contracts.ExecutionClassDeterministic},
{key: "llm-validator", class: contracts.ExecutionClassLLMBacked},
} {
if err := pipeline.RegisterTypedValidatorBuilder(catalog.Validators, effectiveArtifactKind, pipeline.ValidatorSpec{Key: validator.key, ExecutionClass: validator.class}, func(options map[string]any) error {
return pipeline.RejectUnknownOptions(options, "nested")
}, func(pipeline.BuildRequest) (contracts.TypedValidator[effectiveArtifact], error) {
return effectiveValidator{name: validator.key, class: validator.class}, nil
}); err != nil {
t.Fatal(err)
}
}
if err := pipeline.RegisterTypedValidatorBuilder(catalog.Validators, effectiveArtifactKind, pipeline.ValidatorSpec{Key: "option-validator", ExecutionClass: contracts.ExecutionClassDeterministic}, func(options map[string]any) error {
return pipeline.RejectUnknownOptions(options, "allowed")
}, func(pipeline.BuildRequest) (contracts.TypedValidator[effectiveArtifact], error) {
return effectiveValidator{name: "option-validator", class: contracts.ExecutionClassDeterministic}, nil
}); err != nil {
t.Fatal(err)
}
if err := catalog.ValidatorChains.Register(pipeline.ValidatorChainMapping{Stage: pipeline.StageExtract, Module: "extract", Validators: []pipeline.ModuleBinding{pipeline.Binding("default-validator")}}); err != nil {
t.Fatal(err)
}
return catalog
}

View File

@@ -0,0 +1,202 @@
package config
import (
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestPrecedenceFileValuesOverrideBuiltInDefaults(t *testing.T) {
cfg := applyFileConfig(t, `version: 3
concurrency:
total_llm: 4
stage_workers:
extract: 2
output:
directory: ./file-output
cache:
chunk_plans:
directory: ./file-plans
mode: refresh
checkpoints:
directory: ./file-checkpoints
debug:
directory: ./file-debug
`)
if cfg.Concurrency.TotalLLM != 4 || cfg.Concurrency.StageWorkers["extract"] != 2 ||
cfg.Output.Directory != "./file-output" || cfg.Cache.ChunkPlans.Directory != "file-plans" ||
cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheRefresh || cfg.Cache.Checkpoints.Directory != "file-checkpoints" ||
cfg.Debug.Directory != "./file-debug" {
t.Fatalf("file values did not override defaults: %#v", cfg)
}
}
func TestPrecedenceOperationalEnvironmentOverridesFileValues(t *testing.T) {
cfg := applyFileConfig(t, `version: 3
concurrency:
total_llm: 2
stage_workers:
extract: 1
output:
directory: ./file-output
cache:
chunk_plans:
directory: ./file-plans
mode: refresh
checkpoints:
directory: ./file-checkpoints
debug:
directory: ./file-debug
`)
env := map[string]string{
"NOTARIUS_TOTAL_LLM_CONCURRENCY": "8",
"NOTARIUS_STAGE_WORKERS_EXTRACT": "6",
"NOTARIUS_OUTPUT_DIR": "/env/output",
"NOTARIUS_CACHE_CHUNK_PLANS_MODE": "bypass",
"NOTARIUS_CACHE_CHUNK_PLANS_DIR": "/env/plans",
"NOTARIUS_CACHE_CHECKPOINTS_DIR": "/env/checkpoints",
"NOTARIUS_DEBUG_DIR": "/env/debug",
}
if err := cfg.ApplyEnvOverridesWithLookup(lookupValues(env)); err != nil {
t.Fatal(err)
}
if cfg.Concurrency.TotalLLM != 8 || cfg.Concurrency.StageWorkers["extract"] != 6 ||
cfg.Output.Directory != "/env/output" || cfg.Cache.ChunkPlans.Directory != "/env/plans" ||
cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheBypass || cfg.Cache.Checkpoints.Directory != "/env/checkpoints" ||
cfg.Debug.Directory != "/env/debug" {
t.Fatalf("environment values did not override file values: %#v", cfg)
}
}
func TestPrecedenceExtractWorkersFollowEffectiveConcurrencyUnlessExplicit(t *testing.T) {
tests := []struct {
name string
file string
env map[string]string
wantTotal int
wantWorker int
}{
{
name: "default follows environment total",
file: "version: 3\n",
env: map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "5"},
wantTotal: 5,
wantWorker: 5,
},
{
name: "file worker is retained",
file: "version: 3\nconcurrency:\n total_llm: 3\n stage_workers:\n extract: 2\n",
env: map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6"},
wantTotal: 6,
wantWorker: 2,
},
{
name: "environment worker is retained",
file: "version: 3\nconcurrency:\n total_llm: 2\n",
env: map[string]string{
"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6",
"NOTARIUS_STAGE_WORKERS_EXTRACT": "4",
},
wantTotal: 6,
wantWorker: 4,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := applyFileConfig(t, tt.file)
if err := cfg.ApplyEnvOverridesWithLookup(lookupValues(tt.env)); err != nil {
t.Fatal(err)
}
if cfg.Concurrency.TotalLLM != tt.wantTotal || cfg.Concurrency.StageWorkers["extract"] != tt.wantWorker {
t.Fatalf("concurrency = %#v, want total %d and extract %d", cfg.Concurrency, tt.wantTotal, tt.wantWorker)
}
})
}
}
func TestPrecedenceEmptyFileCacheDirectoriesDeferPerUserResolution(t *testing.T) {
cfg := applyFileConfig(t, `version: 3
cache:
chunk_plans:
directory: ""
checkpoints:
directory: ""
`)
if err := cfg.Validate(); err != nil {
t.Fatalf("empty file cache directories should be valid: %v", err)
}
if cfg.Cache.ChunkPlans.Directory != "" || cfg.Cache.Checkpoints.Directory != "" {
t.Fatalf("empty cache directories were not preserved for deferred resolution: %#v", cfg.Cache)
}
resolver := func() (string, error) { return "/user/cache", nil }
chunkPlans, err := DefaultChunkPlanRoot(resolver)
if err != nil {
t.Fatal(err)
}
checkpoints, err := DefaultCheckpointRoot(resolver)
if err != nil {
t.Fatal(err)
}
if chunkPlans != "/user/cache/notarius/chunk-plans" || checkpoints != "/user/cache/notarius/checkpoints" {
t.Fatalf("deferred cache roots = %q, %q", chunkPlans, checkpoints)
}
}
func TestEnvEmptyDirectoryOverridesAreErrors(t *testing.T) {
tests := []string{
"NOTARIUS_OUTPUT_DIR",
"NOTARIUS_CACHE_CHUNK_PLANS_DIR",
"NOTARIUS_CACHE_CHECKPOINTS_DIR",
"NOTARIUS_DEBUG_DIR",
}
for _, name := range tests {
t.Run(name, func(t *testing.T) {
cfg := Default()
err := cfg.ApplyEnvOverridesWithLookup(lookupValues(map[string]string{name: " \t"}))
if err == nil || !strings.Contains(err.Error(), name) {
t.Fatalf("error = %v, want responsible environment variable", err)
}
})
}
}
func TestEnvInvalidIntegersAndChunkCacheModesReportTheirNames(t *testing.T) {
tests := map[string]string{
"NOTARIUS_TOTAL_LLM_CONCURRENCY": "not-an-integer",
"NOTARIUS_STAGE_WORKERS_EXTRACT": "not-an-integer",
"NOTARIUS_CACHE_CHUNK_PLANS_MODE": "not-a-cache-mode",
}
for name, value := range tests {
t.Run(name, func(t *testing.T) {
cfg := Default()
err := cfg.ApplyEnvOverridesWithLookup(lookupValues(map[string]string{name: value}))
if err == nil || !strings.Contains(err.Error(), name) {
t.Fatalf("error = %v, want responsible environment variable", err)
}
})
}
}
func TestEnvRemovedProviderVariablesAreIgnored(t *testing.T) {
before := Default()
cfg := Default()
removed := map[string]string{
"NOTARIUS_LLM_DEFAULT_ENDPOINT": "ignored-provider-setting",
"NOTARIUS_LLM_DEFAULT_MODEL": "ignored-provider-setting",
}
if err := cfg.ApplyEnvOverridesWithLookup(lookupValues(removed)); err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(cfg, before) {
t.Fatalf("removed provider variables changed configuration: %#v", cfg)
}
}
func lookupValues(values map[string]string) func(string) (string, bool) {
return func(name string) (string, bool) {
value, ok := values[name]
return value, ok
}
}

View File

@@ -117,6 +117,63 @@ func TestRedactedEffectiveConfigPayloadDoesNotAliasSource(t *testing.T) {
assertRedactionTestBindingUnchanged(t, effective.ResolvedPipeline.Input, "effective")
}
func TestRedactedSummaryPayloadsCoverEveryEffectiveConfigBinding(t *testing.T) {
bindings := map[string]pipeline.ModuleBinding{}
for _, name := range []string{"input", "chunk", "output", "extract", "merge", "normalize", "lane-validator"} {
bindings[name] = redactionTestBinding("summary-" + name)
}
effective := EffectiveConfig{
Config: Config{Pipelines: map[string]pipeline.PipelineProfile{
"redaction-test": {
Input: bindings["input"],
Chunk: bindings["chunk"],
Output: bindings["output"],
Artifacts: map[string]pipeline.ArtifactLaneProfile{
"safe-lane": {
Extract: bindings["extract"],
Merge: bindings["merge"],
Normalize: bindings["normalize"],
Validators: []pipeline.ModuleBinding{bindings["lane-validator"]},
},
},
},
}},
ResolvedPipeline: pipeline.ResolvedPipeline{
Input: bindings["input"],
Chunk: bindings["chunk"],
Output: bindings["output"],
ArtifactLanes: []pipeline.ResolvedArtifactLane{{
ID: "safe-lane",
Extract: bindings["extract"],
Merge: bindings["merge"],
Normalize: bindings["normalize"],
Validators: []pipeline.ModuleBinding{bindings["lane-validator"]},
}},
},
}
payload, ok := effective.RedactedSummaryPayload().(EffectiveConfig)
if !ok {
t.Fatal("RedactedSummaryPayload() returned an unexpected type")
}
encoded, err := json.Marshal(payload)
if err != nil {
t.Fatal(err)
}
text := string(encoded)
for name := range bindings {
if strings.Contains(text, "summary-"+name+"-secret") || strings.Contains(text, "summary-"+name+"-nested-secret") {
t.Fatalf("summary payload contains sensitive option for %q: %s", name, text)
}
if !strings.Contains(text, "summary-"+name+"-safe") {
t.Fatalf("summary payload omitted safe option for %q: %s", name, text)
}
}
if !strings.Contains(text, "[REDACTED]") {
t.Fatalf("summary payload contains no redaction marker: %s", text)
}
}
func TestRedactedResolvedPipelinePayloadHandlesTypedOptionContainers(t *testing.T) {
type optionMap map[string]string
type optionList []optionMap

View File

@@ -106,11 +106,16 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) erro
if err := validateReferenceMap(id, "", profile.References); err != nil {
return err
}
seenLanes := make(map[string]struct{}, len(profile.Artifacts))
for rawLaneID, lane := range profile.Artifacts {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return fmt.Errorf("pipeline %q artifact lane id must not be empty", id)
}
if _, ok := seenLanes[laneID]; ok {
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated after trimming", id, laneID)
}
seenLanes[laneID] = struct{}{}
if err := validateReferenceMap(id, laneID, lane.References); err != nil {
return err
}

View File

@@ -0,0 +1,446 @@
package config
import (
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestValidateConcurrencyRules(t *testing.T) {
tests := []struct {
name string
setup func(*Config)
want string
}{
{
name: "non-positive total",
setup: func(cfg *Config) {
cfg.Concurrency.TotalLLM = 0
},
want: "total LLM concurrency must be greater than zero",
},
{
name: "worker below one",
setup: func(cfg *Config) {
cfg.Concurrency.TotalLLM = 3
cfg.Concurrency.StageWorkers = map[string]int{"extract": 0}
cfg.Concurrency.extractWorkersConfigured = true
},
want: "stage_workers.extract must be between 1",
},
{
name: "worker above total",
setup: func(cfg *Config) {
cfg.Concurrency.TotalLLM = 3
cfg.Concurrency.StageWorkers = map[string]int{"extract": 4}
cfg.Concurrency.extractWorkersConfigured = true
},
want: "stage_workers.extract must be between 1",
},
{
name: "worker lower boundary",
setup: func(cfg *Config) {
cfg.Concurrency.TotalLLM = 3
cfg.Concurrency.StageWorkers = map[string]int{"extract": 1}
cfg.Concurrency.extractWorkersConfigured = true
},
},
{
name: "worker upper boundary",
setup: func(cfg *Config) {
cfg.Concurrency.TotalLLM = 3
cfg.Concurrency.StageWorkers = map[string]int{"extract": 3}
cfg.Concurrency.extractWorkersConfigured = true
},
},
{
name: "unknown worker key",
setup: func(cfg *Config) {
cfg.Concurrency.StageWorkers = map[string]int{"worker": 1}
},
want: "stage_workers key \"worker\" is not supported",
},
{
name: "blank worker key",
setup: func(cfg *Config) {
cfg.Concurrency.StageWorkers = map[string]int{" ": 1}
},
want: "stage_workers key must not be empty",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Default()
tt.setup(&cfg)
err := cfg.Validate()
if tt.want == "" {
if err != nil {
t.Fatalf("Validate() error = %v", err)
}
return
}
if err == nil || !strings.Contains(err.Error(), tt.want) {
t.Fatalf("Validate() error = %v, want context %q", err, tt.want)
}
})
}
}
func TestValidateScriptoriumSourcesAreMutuallyExclusive(t *testing.T) {
cfg := Default()
cfg.Scriptorium = ScriptoriumConfig{ProfileDir: "./profiles", ProfileFile: "./profile.yml"}
assertValidationContains(t, cfg, "scriptorium profile_dir and profile_file are mutually exclusive")
}
func TestValidateStateSurfaceRules(t *testing.T) {
tests := []struct {
name string
setup func(*Config)
want string
}{
{
name: "blank output root",
setup: func(cfg *Config) {
cfg.Output.Directory = " "
},
want: "output.directory must not be empty",
},
{
name: "blank debug root",
setup: func(cfg *Config) {
cfg.Debug.Directory = " "
},
want: "debug.directory must not be empty",
},
{
name: "NUL in output root",
setup: func(cfg *Config) {
cfg.Output.Directory = "./out\x00put"
},
want: "output.directory must not contain NUL",
},
{
name: "NUL in chunk plan root",
setup: func(cfg *Config) {
cfg.Cache.ChunkPlans.Directory = "./plans\x00"
},
want: "cache.chunk_plans.directory must not contain NUL",
},
{
name: "NUL in checkpoint root",
setup: func(cfg *Config) {
cfg.Cache.Checkpoints.Directory = "./checkpoints\x00"
},
want: "cache.checkpoints.directory must not contain NUL",
},
{
name: "NUL in debug root",
setup: func(cfg *Config) {
cfg.Debug.Directory = "./debug\x00"
},
want: "debug.directory must not contain NUL",
},
{
name: "invalid chunk plan mode",
setup: func(cfg *Config) {
cfg.Cache.ChunkPlans.Mode = pipeline.ChunkCacheMode("invalid")
},
want: "cache.chunk_plans.mode:",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Default()
tt.setup(&cfg)
assertValidationContains(t, cfg, tt.want)
})
}
}
func TestValidateIdentifiersAfterTrimming(t *testing.T) {
tests := []struct {
name string
setup func(*Config)
want string
}{
{
name: "empty pipeline id",
setup: func(cfg *Config) {
cfg.Pipelines = map[string]pipeline.PipelineProfile{" ": {}}
},
want: "pipeline id must not be empty",
},
{
name: "duplicate pipeline ids",
setup: func(cfg *Config) {
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": {}, " main ": {}}
},
want: "pipeline id \"main\" is duplicated after trimming",
},
{
name: "empty lane id",
setup: func(cfg *Config) {
profile := validationProfile()
profile.Artifacts = map[string]pipeline.ArtifactLaneProfile{" ": {}}
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "artifact lane id must not be empty",
},
{
name: "duplicate lane ids",
setup: func(cfg *Config) {
profile := validationProfile()
profile.Artifacts = map[string]pipeline.ArtifactLaneProfile{"spells": {}, " spells ": {}}
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "artifact lane id \"spells\" is duplicated after trimming",
},
{
name: "empty reference slot",
setup: func(cfg *Config) {
profile := validationProfile()
profile.References = map[string]string{" ": "source.txt"}
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "reference slot name must not be empty",
},
{
name: "duplicate reference slots",
setup: func(cfg *Config) {
profile := validationProfile()
profile.References = map[string]string{"slot": "one.txt", " slot ": "two.txt"}
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "reference slot \"slot\" is duplicated after trimming",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Default()
tt.setup(&cfg)
assertValidationContains(t, cfg, tt.want)
})
}
}
func TestValidateBindingRetriesAndProfiles(t *testing.T) {
tests := []struct {
name string
setup func(*Config)
want string
}{
{
name: "negative retries",
setup: func(cfg *Config) {
profile := validationProfile()
profile.Input.Retries = -1
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "input retries must be greater than or equal to zero",
},
{
name: "whitespace-only input profile",
setup: func(cfg *Config) {
profile := validationProfile()
profile.Input.LLMProfile = " "
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
},
want: "input llm_profile must not be empty when set",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Default()
tt.setup(&cfg)
assertValidationContains(t, cfg, tt.want)
})
}
}
func TestValidateReferencesAreUnsupportedOnInputAndOutput(t *testing.T) {
tests := []struct {
name string
set func(*pipeline.PipelineProfile)
want string
}{
{
name: "input references",
set: func(profile *pipeline.PipelineProfile) {
profile.Input.References = map[string]string{"slot": "source.txt"}
},
want: "input references are not supported",
},
{
name: "output references",
set: func(profile *pipeline.PipelineProfile) {
profile.Output.References = map[string]string{"slot": "source.txt"}
},
want: "output references are not supported",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
profile := validationProfile()
tt.set(&profile)
cfg := Default()
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
assertValidationContains(t, cfg, tt.want)
})
}
}
func TestValidateValidatorBindingRules(t *testing.T) {
tests := []struct {
name string
setup func(*pipeline.PipelineProfile)
want string
}{
{
name: "empty validator module",
setup: func(profile *pipeline.PipelineProfile) {
profile.Chunk.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{}},
}
},
want: "chunk validators[0] module must not be empty",
},
{
name: "validator retries",
setup: func(profile *pipeline.PipelineProfile) {
profile.Chunk.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "validator",
Retries: 1,
}},
}
},
want: "chunk validators[0] retries are not supported",
},
{
name: "validator references",
setup: func(profile *pipeline.PipelineProfile) {
profile.Chunk.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "validator",
References: map[string]string{"slot": "source.txt"},
}},
}
},
want: "chunk validators[0] references are not supported",
},
{
name: "nested validators",
setup: func(profile *pipeline.PipelineProfile) {
profile.Chunk.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{
Module: "validator",
Validators: pipeline.ValidatorOverride{Set: true},
}},
}
},
want: "chunk validators[0] nested validators are not supported",
},
{
name: "input validator chain",
setup: func(profile *pipeline.PipelineProfile) {
profile.Input.Validators = pipeline.ValidatorOverride{Set: true}
},
want: "input validators are not supported",
},
{
name: "output validator chain",
setup: func(profile *pipeline.PipelineProfile) {
profile.Output.Validators = pipeline.ValidatorOverride{Set: true}
},
want: "output validators are not supported",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
profile := validationProfile()
tt.setup(&profile)
cfg := Default()
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
assertValidationContains(t, cfg, tt.want)
})
}
}
func TestValidateLaneValidatorCompatibility(t *testing.T) {
tests := []struct {
name string
lane func(*pipeline.ArtifactLaneProfile)
want string
}{
{
name: "deprecated non-empty lane validators",
lane: func(lane *pipeline.ArtifactLaneProfile) {
lane.Validators = []pipeline.ModuleBinding{{Module: "old-validator"}}
},
want: "validators are not supported at artifact lane level",
},
{
name: "stage validators omitted",
lane: func(lane *pipeline.ArtifactLaneProfile) {
},
},
{
name: "stage validators explicitly empty",
lane: func(lane *pipeline.ArtifactLaneProfile) {
lane.Extract.Validators = pipeline.ValidatorOverride{Set: true}
},
},
{
name: "stage validators configured",
lane: func(lane *pipeline.ArtifactLaneProfile) {
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{{Module: "validator"}},
}
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
profile := validationProfile()
lane := profile.Artifacts["lane"]
tt.lane(&lane)
profile.Artifacts["lane"] = lane
cfg := Default()
cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
err := cfg.Validate()
if tt.want == "" {
if err != nil {
t.Fatalf("Validate() error = %v", err)
}
return
}
if err == nil || !strings.Contains(err.Error(), tt.want) {
t.Fatalf("Validate() error = %v, want context %q", err, tt.want)
}
})
}
}
func validationProfile() pipeline.PipelineProfile {
return pipeline.PipelineProfile{
ID: "main",
Input: pipeline.Binding("input"),
Artifacts: map[string]pipeline.ArtifactLaneProfile{
"lane": {},
},
}
}
func assertValidationContains(t *testing.T, cfg Config, want string) {
t.Helper()
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), want) {
t.Fatalf("Validate() error = %v, want context %q", err, want)
}
}