Add Seriatim pipeline config coverage

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{}
+)

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