notarius/internal/modules/seriatim/input/transcript/config_test.go

package transcript

import (
	"context"
	"encoding/json"
	"os"
	"reflect"
	"strings"
	"testing"

	"gitea.maximumdirect.net/eric/notarius/internal/core/config"
	"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 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,
		ArtifactKind: seriatimArtifactKind,
		Requires:     []string{"chunks", "transcript.speaker", "transcript.timestamps"},
		Provides:     []string{"fake.artifacts"},
	})
	mustRegisterMerger(t, mergers, pipeline.ModuleSpec{
		Key:          pipeline.DefaultMergeModule,
		Stage:        pipeline.StageMerge,
		ArtifactKind: seriatimArtifactKind,
		Requires:     []string{"fake.artifacts"},
	})
	mustRegisterNormalizer(t, normalizers, pipeline.ModuleSpec{
		Key:          pipeline.DefaultNormalizeModule,
		Stage:        pipeline.StageNormalize,
		ArtifactKind: seriatimArtifactKind,
	})
	mustRegisterOutput(t, outputs, pipeline.ModuleSpec{
		Key:   pipeline.DefaultOutputModule,
		Stage: pipeline.StageOutput,
	})

	codecs := pipeline.NewArtifactCodecRegistry()
	if err := pipeline.RegisterArtifactCodec(codecs, seriatimArtifactCodec{}); err != nil {
		t.Fatalf("register artifact codec: %v", err)
	}
	return pipeline.ModuleCatalog{
		Inputs:          inputs,
		Chunkers:        chunkers,
		ArtifactCodecs:  codecs,
		Extractors:      extractors,
		Mergers:         mergers,
		Normalizers:     normalizers,
		ValidatorChains: pipeline.NewValidatorChainRegistry(),
		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 := pipeline.RegisterExtractor[seriatimArtifact](registry, spec, func() (contracts.Extractor[seriatimArtifact], 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 := pipeline.RegisterMerger[seriatimArtifact](registry, spec, func() (contracts.Merger[seriatimArtifact], error) {
		return fakeMerger{}, nil
	}); err != nil {
		t.Fatalf("register merger: %v", err)
	}
}

func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry, spec pipeline.ModuleSpec) {
	t.Helper()
	if err := pipeline.RegisterNormalizer[seriatimArtifact](registry, spec, func() (contracts.Normalizer[seriatimArtifact], error) {
		return fakeNormalizer{}, 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) ReferenceSlots() []contracts.ReferenceSlot { return nil }

func (fakeChunker) Plan(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
	return contracts.ChunkPlanResult{}, nil
}

type fakeExtractor struct{}

func (fakeExtractor) Key() string { return "fake/extract" }

func (fakeExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil }

func (fakeExtractor) Extract(ctx context.Context, req contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[seriatimArtifact], error) {
	return contracts.TypedExtractionResult[seriatimArtifact]{}, nil
}

type fakeMerger struct{}

func (fakeMerger) Key() string { return pipeline.DefaultMergeModule }

func (fakeMerger) Merge(ctx context.Context, req contracts.TypedMergeRequest[seriatimArtifact]) (contracts.TypedMergeResult[seriatimArtifact], error) {
	if len(req.ExtractOutputs) == 0 {
		return contracts.TypedMergeResult[seriatimArtifact]{}, nil
	}
	return contracts.TypedMergeResult[seriatimArtifact]{Value: req.ExtractOutputs[0].Value}, nil
}

type fakeNormalizer struct{}

func (fakeNormalizer) Key() string { return pipeline.DefaultNormalizeModule }

func (fakeNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }

func (fakeNormalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[seriatimArtifact]) (contracts.TypedNormalizeResult[seriatimArtifact], error) {
	return contracts.TypedNormalizeResult[seriatimArtifact]{Value: req.MergeOutput.Value}, 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[seriatimArtifact] = fakeExtractor{}
	_ contracts.OutputEncoder               = fakeOutput{}
)

const seriatimArtifactKind contracts.ArtifactKind = "test/seriatim-event"

type seriatimArtifact struct {
	Value      string             `json:"value"`
	SourceRefs []source.SourceRef `json:"source_refs"`
}

type seriatimArtifactCodec struct{}

func (seriatimArtifactCodec) Kind() contracts.ArtifactKind { return seriatimArtifactKind }
func (seriatimArtifactCodec) Schema() contracts.ArtifactSchema {
	return contracts.ArtifactSchema{ID: "fake.event", Name: "fake_event", Version: "v1", JSONSchema: []byte(`{"type":"object"}`)}
}
func (seriatimArtifactCodec) MediaType() string { return "application/json" }
func (seriatimArtifactCodec) EncodeCandidate(value seriatimArtifact) ([]byte, error) {
	return json.Marshal(value)
}
func (seriatimArtifactCodec) Encode(value seriatimArtifact) ([]byte, error) {
	return json.Marshal(value)
}
func (seriatimArtifactCodec) Decode(content []byte) (seriatimArtifact, error) {
	var value seriatimArtifact
	err := json.Unmarshal(content, &value)
	return value, err
}