From ef8bd91d48b6f09fcf70675c5cd784c4a90bfda9 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 3 Jul 2026 21:15:06 +0000 Subject: [PATCH] Add Seriatim pipeline config coverage --- .../modules/input/seriatim/config_test.go | 250 ++++++++++++++++++ .../input/seriatim/testdata/pipeline.yml | 11 + 2 files changed, 261 insertions(+) create mode 100644 internal/modules/input/seriatim/config_test.go create mode 100644 internal/modules/input/seriatim/testdata/pipeline.yml diff --git a/internal/modules/input/seriatim/config_test.go b/internal/modules/input/seriatim/config_test.go new file mode 100644 index 00000000..acf259d2 --- /dev/null +++ b/internal/modules/input/seriatim/config_test.go @@ -0,0 +1,250 @@ +package seriatim + +import ( + "context" + "os" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/config" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +func TestPipelineConfigLoadsAndResolvesWithSeriatimInput(t *testing.T) { + cfg := loadPipelineConfig(t) + + resolved, err := cfg.Resolve(config.ResolveInput{ + PipelineID: "seriatim-fixture", + Catalog: seriatimTestCatalog(t, ModuleSpec()), + }) + if err != nil { + t.Fatalf("Resolve() error = %v, want nil", err) + } + + if resolved.ResolvedPipeline.Input.Module != Key { + t.Fatalf("resolved input module = %q, want %q", resolved.ResolvedPipeline.Input.Module, Key) + } + if resolved.ResolvedPipeline.Digest == "" { + t.Fatal("resolved digest is empty") + } + + again, err := cfg.Resolve(config.ResolveInput{ + PipelineID: "seriatim-fixture", + Catalog: seriatimTestCatalog(t, ModuleSpec()), + }) + if err != nil { + t.Fatalf("second Resolve() error = %v, want nil", err) + } + if resolved.ResolvedPipeline.Digest != again.ResolvedPipeline.Digest { + t.Fatalf("resolved digest = %q, second digest = %q; want stable digest", resolved.ResolvedPipeline.Digest, again.ResolvedPipeline.Digest) + } +} + +func TestPipelineConfigRejectsMissingSeriatimCapability(t *testing.T) { + spec := ModuleSpec() + spec.Provides = withoutCapability(spec.Provides, "transcript.timestamps") + cfg := loadPipelineConfig(t) + + _, err := cfg.Resolve(config.ResolveInput{ + PipelineID: "seriatim-fixture", + Catalog: seriatimTestCatalog(t, spec), + }) + if err == nil { + t.Fatal("Resolve() error = nil, want missing capability error") + } + if !strings.Contains(err.Error(), "missing capability") || !strings.Contains(err.Error(), "transcript.timestamps") { + t.Fatalf("Resolve() error = %q, want missing transcript.timestamps capability", err.Error()) + } +} + +func TestPipelineConfigRejectsUnknownLaneSelection(t *testing.T) { + cfg := loadPipelineConfig(t) + + _, err := cfg.Resolve(config.ResolveInput{ + PipelineID: "seriatim-fixture", + Only: []string{"missing"}, + Catalog: seriatimTestCatalog(t, ModuleSpec()), + }) + if err == nil { + t.Fatal("Resolve() error = nil, want unknown lane error") + } + if !strings.Contains(err.Error(), "selected artifact lane") || !strings.Contains(err.Error(), "missing") { + t.Fatalf("Resolve() error = %q, want unknown lane context", err.Error()) + } +} + +func loadPipelineConfig(t *testing.T) config.Config { + t.Helper() + + data, err := os.ReadFile("testdata/pipeline.yml") + if err != nil { + t.Fatalf("ReadFile(pipeline.yml) error = %v, want nil", err) + } + fileCfg, err := config.ParseFileConfigYAML(data) + if err != nil { + t.Fatalf("ParseFileConfigYAML() error = %v, want nil", err) + } + + cfg := config.Default() + if err := cfg.ApplyFileConfig(fileCfg); err != nil { + t.Fatalf("ApplyFileConfig() error = %v, want nil", err) + } + + profile, ok := cfg.Pipelines["seriatim-fixture"] + if !ok { + t.Fatal("pipeline seriatim-fixture was not loaded") + } + if profile.Input.Module != Key { + t.Fatalf("loaded input module = %q, want %q", profile.Input.Module, Key) + } + return cfg +} + +func seriatimTestCatalog(t *testing.T, inputSpec pipeline.ModuleSpec) pipeline.ModuleCatalog { + t.Helper() + + inputs := pipeline.NewInputAdapterRegistry() + chunkers := pipeline.NewChunkerRegistry() + extractors := pipeline.NewExtractorRegistry() + mergers := pipeline.NewMergerRegistry() + normalizers := pipeline.NewNormalizerRegistry() + outputs := pipeline.NewOutputEncoderRegistry() + + if reflect.DeepEqual(inputSpec, ModuleSpec()) { + if err := Register(inputs); err != nil { + t.Fatalf("register seriatim input: %v", err) + } + } else if err := inputs.RegisterWithSpec(inputSpec, func() (contracts.InputAdapter, error) { + return New(), nil + }); err != nil { + t.Fatalf("register seriatim input override: %v", err) + } + + mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{ + Key: "fake/chunk", + Stage: pipeline.StageChunk, + Requires: []string{"source.transcript"}, + Provides: []string{"chunks"}, + }) + mustRegisterExtractor(t, extractors, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks", "transcript.speaker", "transcript.timestamps"}, + Provides: []string{"fake.artifacts"}, + }) + mustRegisterMerger(t, mergers, pipeline.ModuleSpec{ + Key: pipeline.DefaultMergeModule, + Stage: pipeline.StageMerge, + Requires: []string{"fake.artifacts"}, + }) + mustRegisterNormalizer(t, normalizers, pipeline.ModuleSpec{ + Key: pipeline.DefaultNormalizeModule, + Stage: pipeline.StageNormalize, + }) + mustRegisterOutput(t, outputs, pipeline.ModuleSpec{ + Key: pipeline.DefaultOutputModule, + Stage: pipeline.StageOutput, + }) + + return pipeline.ModuleCatalog{ + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + Outputs: outputs, + } +} + +func mustRegisterChunker(t *testing.T, registry *pipeline.ChunkerRegistry, spec pipeline.ModuleSpec) { + t.Helper() + if err := registry.RegisterWithSpec(spec, func() (contracts.Chunker, error) { + return fakeChunker{}, 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 fakeExtractor{}, 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 pipeline.AppendOrderMerger{}, 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 pipeline.NoopNormalizer{}, nil + }); err != nil { + t.Fatalf("register normalizer: %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 fakeOutput{}, nil + }); err != nil { + t.Fatalf("register output: %v", err) + } +} + +type fakeChunker struct{} + +func (fakeChunker) Key() string { return "fake/chunk" } + +func (fakeChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) { + return contracts.ChunkResult{}, nil +} + +type fakeExtractor struct{} + +func (fakeExtractor) Key() string { return "fake/extract" } + +func (fakeExtractor) ArtifactType() string { return "fake" } + +func (fakeExtractor) SchemaVersion() string { return "v1" } + +func (fakeExtractor) Validators() []contracts.Validator { return nil } + +func (fakeExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) { + return contracts.ExtractionResult{}, nil +} + +type fakeOutput struct{} + +func (fakeOutput) Key() string { return pipeline.DefaultOutputModule } + +func (fakeOutput) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) { + return contracts.OutputResult{}, nil +} + +func withoutCapability(capabilities []string, capability string) []string { + filtered := make([]string, 0, len(capabilities)) + for _, candidate := range capabilities { + if candidate != capability { + filtered = append(filtered, candidate) + } + } + return filtered +} + +var ( + _ contracts.Chunker = fakeChunker{} + _ contracts.Extractor = fakeExtractor{} + _ contracts.OutputEncoder = fakeOutput{} +) diff --git a/internal/modules/input/seriatim/testdata/pipeline.yml b/internal/modules/input/seriatim/testdata/pipeline.yml new file mode 100644 index 00000000..be823d8a --- /dev/null +++ b/internal/modules/input/seriatim/testdata/pipeline.yml @@ -0,0 +1,11 @@ +version: 1 +pipelines: + seriatim-fixture: + input: seriatim + chunk: fake/chunk + artifacts: + events: + extract: fake/extract + merge: appendorder + normalize: noop + output: json