diff --git a/internal/core/config/effective_config.go b/internal/core/config/effective_config.go new file mode 100644 index 0000000..8de750a --- /dev/null +++ b/internal/core/config/effective_config.go @@ -0,0 +1,96 @@ +package config + +import ( + "fmt" + "strings" + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/llm" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +type ResolveInput struct { + PipelineID string + Only []string + Catalog pipeline.ModuleCatalog +} + +type EffectiveConfig struct { + Config Config + PipelineID string + Only []string + ResolvedPipeline pipeline.ResolvedPipeline +} + +func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { + if err := c.Validate(); err != nil { + return EffectiveConfig{}, err + } + + pipelineID := strings.TrimSpace(input.PipelineID) + if pipelineID == "" { + return EffectiveConfig{}, fmt.Errorf("pipeline id must not be empty") + } + + profile, ok := lookupPipelineProfile(c.Pipelines, pipelineID) + if !ok { + return EffectiveConfig{}, fmt.Errorf("pipeline %q is not configured", pipelineID) + } + profile = clonePipelineProfile(profile) + profile.ID = pipelineID + + resolved, err := pipeline.ResolvePipeline(profile, pipeline.ResolveOptions{Only: input.Only}, input.Catalog) + if err != nil { + return EffectiveConfig{}, fmt.Errorf("resolve pipeline %q: %w", pipelineID, err) + } + + return EffectiveConfig{ + Config: cloneConfig(c), + PipelineID: pipelineID, + Only: append([]string(nil), input.Only...), + ResolvedPipeline: resolved, + }, nil +} + +func lookupPipelineProfile(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 (c Config) OpenAICompatibleClientConfig(profileID string) (llm.OpenAICompatibleClientConfig, error) { + trimmedID := strings.TrimSpace(profileID) + profile, ok := c.LLMProfile(trimmedID) + if !ok { + return llm.OpenAICompatibleClientConfig{}, fmt.Errorf("LLM profile %q is not configured", trimmedID) + } + + provider := strings.TrimSpace(profile.Provider) + if provider == "" { + provider = providerOpenAICompatible + } + if provider != providerOpenAICompatible { + return llm.OpenAICompatibleClientConfig{}, fmt.Errorf("LLM profile %q provider %q is not supported", trimmedID, provider) + } + + baseURL := strings.TrimSpace(profile.BaseURL) + if baseURL == "" { + return llm.OpenAICompatibleClientConfig{}, fmt.Errorf("LLM profile %q base URL must not be empty", trimmedID) + } + model := strings.TrimSpace(profile.Model) + if model == "" { + return llm.OpenAICompatibleClientConfig{}, fmt.Errorf("LLM profile %q model must not be empty", trimmedID) + } + + return llm.OpenAICompatibleClientConfig{ + BaseURL: baseURL, + Model: model, + APIKey: profile.APIKey, + MaxRetries: profile.MaxRetries, + RequestTimeout: time.Duration(profile.TimeoutSeconds) * time.Second, + }, nil +} diff --git a/internal/core/config/effective_config_test.go b/internal/core/config/effective_config_test.go new file mode 100644 index 0000000..89ceb0b --- /dev/null +++ b/internal/core/config/effective_config_test.go @@ -0,0 +1,169 @@ +package config + +import ( + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +func TestResolveRejectsEmptyAndUnknownPipelineID(t *testing.T) { + tests := []struct { + name string + pipelineID string + want string + }{ + {name: "empty", pipelineID: " ", want: "pipeline id"}, + {name: "unknown", pipelineID: "missing", want: "not configured"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, err := validConfig().Resolve(ResolveInput{PipelineID: tc.pipelineID, Catalog: fakeCatalog(t)}) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("expected error containing %q, got %v", tc.want, err) + } + }) + } +} + +func TestResolveLaneFilteringSuccessAndFailure(t *testing.T) { + effective, err := validConfig().Resolve(ResolveInput{ + PipelineID: " example ", + Only: []string{" notes "}, + Catalog: fakeCatalog(t), + }) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + + if effective.PipelineID != "example" { + t.Fatalf("unexpected pipeline ID: %q", effective.PipelineID) + } + if len(effective.ResolvedPipeline.ArtifactLanes) != 1 || effective.ResolvedPipeline.ArtifactLanes[0].ID != "notes" { + t.Fatalf("unexpected resolved lanes: %+v", effective.ResolvedPipeline.ArtifactLanes) + } + if effective.ResolvedPipeline.Digest == "" { + t.Fatalf("expected digest") + } + + _, err = validConfig().Resolve(ResolveInput{ + PipelineID: "example", + Only: []string{"missing"}, + Catalog: fakeCatalog(t), + }) + if err == nil || !strings.Contains(err.Error(), "selected artifact lane") { + t.Fatalf("expected invalid lane error, got %v", err) + } +} + +func TestResolveUsesTrimmedPipelineMapKeys(t *testing.T) { + cfg := validConfig() + cfg.Pipelines[" example "] = cfg.Pipelines["example"] + delete(cfg.Pipelines, "example") + + effective, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)}) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if effective.PipelineID != "example" { + t.Fatalf("unexpected pipeline ID: %q", effective.PipelineID) + } +} + +func TestResolveSurfacesUnknownModuleKeyThroughCatalog(t *testing.T) { + cfg := validConfig() + lane := cfg.Pipelines["example"].Artifacts["events"] + lane.Extract = pipeline.Binding("missing/extract") + cfg.Pipelines["example"].Artifacts["events"] = lane + + _, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)}) + if err == nil || !strings.Contains(err.Error(), "missing/extract") || !strings.Contains(err.Error(), "events") { + t.Fatalf("expected unknown module error with lane context, got %v", err) + } +} + +func TestResolveSurfacesMissingCapabilityThroughCatalog(t *testing.T) { + _, err := validConfig().Resolve(ResolveInput{ + PipelineID: "example", + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "json", + Stage: pipeline.StageOutput, + Requires: []string{"missing-capability"}, + }), + }) + if err == nil || !strings.Contains(err.Error(), "missing capability") || !strings.Contains(err.Error(), "json") { + t.Fatalf("expected missing capability error, got %v", err) + } +} + +func TestResolveDigestChangesWhenEffectiveConfigChanges(t *testing.T) { + cfg := validConfig() + first, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)}) + if err != nil { + t.Fatalf("Resolve first: %v", err) + } + + lane := cfg.Pipelines["example"].Artifacts["events"] + lane.Extract.Options = map[string]any{"temperature": 0.2} + cfg.Pipelines["example"].Artifacts["events"] = lane + second, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)}) + if err != nil { + t.Fatalf("Resolve second: %v", err) + } + + if first.ResolvedPipeline.Digest == second.ResolvedPipeline.Digest { + t.Fatalf("expected digest to change, got %q", first.ResolvedPipeline.Digest) + } +} + +func TestOpenAICompatibleClientConfigRejectsIncompleteDefaultProfile(t *testing.T) { + cfg := Default() + + _, err := cfg.OpenAICompatibleClientConfig("default") + if err == nil || !strings.Contains(err.Error(), "base URL") { + t.Fatalf("expected incomplete profile error, got %v", err) + } +} + +func TestOpenAICompatibleClientConfigSuccess(t *testing.T) { + cfg := validConfig() + profile := cfg.LLMProfiles["default"] + profile.APIKey = "secret" + profile.TimeoutSeconds = 45 + profile.MaxRetries = 4 + cfg.LLMProfiles["default"] = profile + + llmCfg, err := cfg.OpenAICompatibleClientConfig(" default ") + if err != nil { + t.Fatalf("OpenAICompatibleClientConfig: %v", err) + } + + if llmCfg.BaseURL != "https://example.invalid/v1" || llmCfg.Model != "test-model" || llmCfg.APIKey != "secret" { + t.Fatalf("unexpected client config strings: %+v", llmCfg) + } + if llmCfg.MaxRetries != 4 { + t.Fatalf("unexpected max retries: %d", llmCfg.MaxRetries) + } + if llmCfg.RequestTimeout != 45*time.Second { + t.Fatalf("unexpected timeout: %s", llmCfg.RequestTimeout) + } +} + +func TestOpenAICompatibleClientConfigRejectsUnknownAndUnsupportedProfiles(t *testing.T) { + _, err := validConfig().OpenAICompatibleClientConfig("missing") + if err == nil || !strings.Contains(err.Error(), "not configured") { + t.Fatalf("expected unknown profile error, got %v", err) + } + + cfg := validConfig() + profile := cfg.LLMProfiles["default"] + profile.Provider = "unsupported" + cfg.LLMProfiles["default"] = profile + + _, err = cfg.OpenAICompatibleClientConfig("default") + if err == nil || !strings.Contains(err.Error(), "provider") { + t.Fatalf("expected unsupported provider error, got %v", err) + } +} diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go new file mode 100644 index 0000000..8c3fde2 --- /dev/null +++ b/internal/core/config/validation.go @@ -0,0 +1,162 @@ +package config + +import ( + "fmt" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const providerOpenAICompatible = "openai-compatible" + +func (c Config) Validate() error { + if err := validateLLMProfiles(c.LLMProfiles); err != nil { + return err + } + if err := validateDiagnostics(c.Diagnostics); err != nil { + return err + } + if c.Concurrency.TotalLLM <= 0 { + return fmt.Errorf("total LLM concurrency must be greater than zero") + } + return validatePipelineProfiles(c.Pipelines, c.LLMProfiles) +} + +func (c Config) LLMProfile(id string) (LLMProfile, bool) { + trimmedID := strings.TrimSpace(id) + for rawID, profile := range c.LLMProfiles { + if strings.TrimSpace(rawID) == trimmedID { + return profile, true + } + } + return LLMProfile{}, false +} + +func validateLLMProfiles(profiles map[string]LLMProfile) error { + seen := make(map[string]struct{}, len(profiles)) + for rawID, profile := range profiles { + id := strings.TrimSpace(rawID) + if id == "" { + return fmt.Errorf("LLM profile id must not be empty") + } + if _, ok := seen[id]; ok { + return fmt.Errorf("LLM profile id %q is duplicated after trimming", id) + } + seen[id] = struct{}{} + + provider := strings.TrimSpace(profile.Provider) + if provider != "" && provider != providerOpenAICompatible { + return fmt.Errorf("LLM profile %q provider %q is not supported", id, provider) + } + if profile.TimeoutSeconds < 0 { + return fmt.Errorf("LLM profile %q timeout seconds must not be negative", id) + } + if profile.MaxRetries < 0 { + return fmt.Errorf("LLM profile %q max retries must not be negative", id) + } + if profile.MaxConcurrency < 0 { + return fmt.Errorf("LLM profile %q max concurrency must not be negative", id) + } + if profile.TimeoutSeconds == 0 && profile.MaxRetries == 0 && profile.MaxConcurrency == 0 { + continue + } + if profile.TimeoutSeconds == 0 { + return fmt.Errorf("LLM profile %q timeout seconds must be greater than zero", id) + } + if profile.MaxConcurrency == 0 { + return fmt.Errorf("LLM profile %q max concurrency must be greater than zero", id) + } + } + return nil +} + +func validateDiagnostics(cfg DiagnosticsConfig) error { + if strings.TrimSpace(cfg.WorkDir) == "" { + return fmt.Errorf("diagnostics work dir must not be empty") + } + switch cfg.Retention { + case "", diagnostics.RetentionAuto, diagnostics.RetentionAlways, diagnostics.RetentionNever: + return nil + default: + return fmt.Errorf("diagnostics retention %q is not supported", cfg.Retention) + } +} + +func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile, llmProfiles map[string]LLMProfile) error { + seen := make(map[string]struct{}, len(profiles)) + for rawID, profile := range profiles { + id := strings.TrimSpace(rawID) + if id == "" { + return fmt.Errorf("pipeline id must not be empty") + } + if _, ok := seen[id]; ok { + return fmt.Errorf("pipeline id %q is duplicated after trimming", id) + } + seen[id] = struct{}{} + + if profile.ID != "" && strings.TrimSpace(profile.ID) != id { + return fmt.Errorf("pipeline %q profile id %q does not match map key", id, profile.ID) + } + if err := validateBindingLLMProfile(id, "", "input", profile.Input, llmProfiles); err != nil { + return err + } + if err := validateBindingLLMProfile(id, "", "chunk", profile.Chunk, llmProfiles); err != nil { + return err + } + if err := validateBindingLLMProfile(id, "", "output", profile.Output, llmProfiles); err != nil { + return err + } + 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 err := validateBindingLLMProfile(id, laneID, "extract", lane.Extract, llmProfiles); err != nil { + return err + } + if err := validateBindingLLMProfile(id, laneID, "merge", lane.Merge, llmProfiles); err != nil { + return err + } + if err := validateBindingLLMProfile(id, laneID, "normalize", lane.Normalize, llmProfiles); err != nil { + return err + } + for i, validator := range lane.Validators { + if err := validateBindingLLMProfile(id, laneID, fmt.Sprintf("validator[%d]", i), validator, llmProfiles); err != nil { + return err + } + } + } + } + return nil +} + +func validateBindingLLMProfile( + pipelineID string, + laneID string, + slot string, + binding pipeline.ModuleBinding, + profiles map[string]LLMProfile, +) error { + profileID := strings.TrimSpace(binding.LLMProfile) + if profileID == "" { + profileID = pipeline.DefaultLLMProfile + } + if hasLLMProfile(profiles, profileID) { + return nil + } + if laneID != "" { + return fmt.Errorf("pipeline %q lane %q %s references unknown LLM profile %q", pipelineID, laneID, slot, profileID) + } + return fmt.Errorf("pipeline %q %s references unknown LLM profile %q", pipelineID, slot, profileID) +} + +func hasLLMProfile(profiles map[string]LLMProfile, profileID string) bool { + profileID = strings.TrimSpace(profileID) + for rawID := range profiles { + if strings.TrimSpace(rawID) == profileID { + return true + } + } + return false +} diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go new file mode 100644 index 0000000..6778fdf --- /dev/null +++ b/internal/core/config/validation_test.go @@ -0,0 +1,334 @@ +package config + +import ( + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +func TestValidateSuccessForValidConfig(t *testing.T) { + cfg := validConfig() + + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate: %v", err) + } +} + +func TestValidateRejectsUnknownLLMProfileReferencedByBinding(t *testing.T) { + cfg := validConfig() + lane := cfg.Pipelines["example"].Artifacts["events"] + lane.Extract.LLMProfile = "missing" + cfg.Pipelines["example"].Artifacts["events"] = lane + + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "unknown LLM profile") || !strings.Contains(err.Error(), "events") { + t.Fatalf("expected unknown LLM profile error with lane context, got %v", err) + } +} + +func TestValidateRejectsInvalidProvider(t *testing.T) { + cfg := validConfig() + profile := cfg.LLMProfiles["default"] + profile.Provider = "unsupported" + cfg.LLMProfiles["default"] = profile + + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "provider") { + t.Fatalf("expected provider error, got %v", err) + } +} + +func TestValidateRejectsInvalidNumericFields(t *testing.T) { + tests := []struct { + name string + mutate func(Config) Config + want string + }{ + { + name: "total concurrency", + mutate: func(cfg Config) Config { + cfg.Concurrency.TotalLLM = 0 + return cfg + }, + want: "total LLM concurrency", + }, + { + name: "timeout", + mutate: func(cfg Config) Config { + profile := cfg.LLMProfiles["default"] + profile.TimeoutSeconds = -1 + cfg.LLMProfiles["default"] = profile + return cfg + }, + want: "timeout", + }, + { + name: "max retries", + mutate: func(cfg Config) Config { + profile := cfg.LLMProfiles["default"] + profile.MaxRetries = -1 + cfg.LLMProfiles["default"] = profile + return cfg + }, + want: "max retries", + }, + { + name: "max concurrency", + mutate: func(cfg Config) Config { + profile := cfg.LLMProfiles["default"] + profile.MaxConcurrency = 0 + cfg.LLMProfiles["default"] = profile + return cfg + }, + want: "max concurrency", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := tc.mutate(validConfig()).Validate() + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("expected error containing %q, got %v", tc.want, err) + } + }) + } +} + +func TestValidateRejectsInvalidDiagnosticsRetention(t *testing.T) { + cfg := validConfig() + cfg.Diagnostics.Retention = diagnostics.RetentionMode("sometimes") + + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "retention") { + t.Fatalf("expected retention error, got %v", err) + } +} + +func TestValidateRejectsEmptyIDs(t *testing.T) { + tests := []struct { + name string + mutate func(Config) Config + want string + }{ + { + name: "LLM profile", + mutate: func(cfg Config) Config { + cfg.LLMProfiles[" "] = LLMProfile{} + return cfg + }, + want: "LLM profile id", + }, + { + name: "pipeline", + mutate: func(cfg Config) Config { + cfg.Pipelines[" "] = pipeline.PipelineProfile{} + return cfg + }, + want: "pipeline id", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := tc.mutate(validConfig()).Validate() + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("expected error containing %q, got %v", tc.want, err) + } + }) + } +} + +func TestValidateRejectsIDsDuplicatedAfterTrimming(t *testing.T) { + tests := []struct { + name string + mutate func(Config) Config + want string + }{ + { + name: "LLM profile", + mutate: func(cfg Config) Config { + cfg.LLMProfiles[" default "] = cfg.LLMProfiles["default"] + return cfg + }, + want: "duplicated", + }, + { + name: "pipeline", + mutate: func(cfg Config) Config { + cfg.Pipelines[" example "] = cfg.Pipelines["example"] + return cfg + }, + want: "duplicated", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := tc.mutate(validConfig()).Validate() + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("expected error containing %q, got %v", tc.want, err) + } + }) + } +} + +func TestValidateUsesTrimmedLLMProfileIDs(t *testing.T) { + cfg := validConfig() + cfg.LLMProfiles[" default "] = cfg.LLMProfiles["default"] + delete(cfg.LLMProfiles, "default") + + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate: %v", err) + } + if _, ok := cfg.LLMProfile("default"); !ok { + t.Fatalf("expected trimmed LLM profile lookup to succeed") + } +} + +func validConfig() Config { + cfg := Default() + profile := cfg.LLMProfiles["default"] + profile.BaseURL = "https://example.invalid/v1" + profile.Model = "test-model" + cfg.LLMProfiles["default"] = profile + cfg.Pipelines["example"] = pipeline.PipelineProfile{ + Input: pipeline.Binding("fake/input"), + Artifacts: map[string]pipeline.ArtifactLaneProfile{ + "events": { + Extract: pipeline.Binding("fake/extract"), + Validators: []pipeline.ModuleBinding{pipeline.Binding("fake/validator")}, + }, + "notes": { + Extract: pipeline.Binding("fake/extract"), + }, + }, + } + return cfg +} + +func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.ModuleCatalog { + t.Helper() + 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"}, + }, + "fake/validator": { + Key: "fake/validator", + Stage: pipeline.StageValidate, + Requires: []string{"normalized"}, + Provides: []string{"validated"}, + }, + "json": { + Key: "json", + Stage: pipeline.StageOutput, + Requires: []string{"normalized"}, + }, + } + for _, override := range overrides { + specs[override.Key] = override + } + + inputs := pipeline.NewInputAdapterRegistry() + chunkers := pipeline.NewChunkerRegistry() + extractors := pipeline.NewExtractorRegistry() + mergers := pipeline.NewMergerRegistry() + normalizers := pipeline.NewNormalizerRegistry() + validators := pipeline.NewValidatorRegistry() + outputs := pipeline.NewOutputEncoderRegistry() + + 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"]) + mustRegisterValidator(t, validators, specs["fake/validator"]) + mustRegisterOutput(t, outputs, specs["json"]) + + return pipeline.ModuleCatalog{ + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + Validators: validators, + Outputs: outputs, + } +} + +func mustRegisterInput(t *testing.T, registry *pipeline.InputAdapterRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.InputAdapter, error) { return nil, nil }); err != nil { + t.Fatalf("register input: %v", err) + } +} + +func mustRegisterChunker(t *testing.T, registry *pipeline.ChunkerRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.Chunker, error) { return nil, nil }); err != nil { + t.Fatalf("register chunker: %v", err) + } +} + +func mustRegisterExtractor(t *testing.T, registry *pipeline.ExtractorRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.Extractor, error) { return nil, nil }); err != nil { + t.Fatalf("register extractor: %v", err) + } +} + +func mustRegisterMerger(t *testing.T, registry *pipeline.MergerRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.Merger, error) { return nil, nil }); err != nil { + t.Fatalf("register merger: %v", err) + } +} + +func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.Normalizer, error) { return nil, nil }); err != nil { + t.Fatalf("register normalizer: %v", err) + } +} + +func mustRegisterValidator(t *testing.T, registry *pipeline.ValidatorRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.Validator, error) { return nil, nil }); err != nil { + t.Fatalf("register validator: %v", err) + } +} + +func mustRegisterOutput(t *testing.T, registry *pipeline.OutputEncoderRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.OutputEncoder, error) { return nil, nil }); err != nil { + t.Fatalf("register output: %v", err) + } +}