diff --git a/internal/cli/contract_test_helpers_test.go b/internal/cli/contract_test_helpers_test.go index 2613ef6..e827ecb 100644 --- a/internal/cli/contract_test_helpers_test.go +++ b/internal/cli/contract_test_helpers_test.go @@ -12,3 +12,5 @@ func replaceRequiredOnce(t *testing.T, input, old, replacement string) string { } return strings.Replace(input, old, replacement, 1) } + +func emptyLookup(string) (string, bool) { return "", false } diff --git a/internal/cli/production_contract_test.go b/internal/cli/production_contract_test.go index c7d4ff8..5aba7cd 100644 --- a/internal/cli/production_contract_test.go +++ b/internal/cli/production_contract_test.go @@ -91,40 +91,15 @@ func TestProductionCatalogCoversMaintainedConfigurations(t *testing.T) { t.Fatalf("catalog validator chain = %#v, want %#v", got, wantChain) } - for _, example := range maintainedExampleFiles(t) { - cfg := loadMaintainedExample(t, example.path) - effective, err := cfg.Resolve(config.ResolveInput{PipelineID: "dnd-session", Catalog: catalog}) - if err != nil { - t.Fatalf("resolve %s: %v", example.name, err) - } - if effective.ResolvedPipeline.Input.Module != "seriatim" || len(effective.ResolvedPipeline.ArtifactLanes) != 1 || effective.ResolvedPipeline.ArtifactLanes[0].ID != "spells" { - t.Fatalf("resolved %s pipeline = %#v, want seriatim and spells", example.name, effective.ResolvedPipeline) - } - } } -func TestDefaultCLICompositionResolvesMaintainedConfigurations(t *testing.T) { - catalog, err := effectiveCatalog(Options{}) - if err != nil { - t.Fatalf("resolve default catalog: %v", err) - } - if isEmptyCatalog(catalog) { - t.Fatal("default catalog is empty") - } - registries, err := effectiveRegistries(Options{}) - if err != nil { - t.Fatalf("resolve default registries: %v", err) - } - if isEmptyRegistries(registries) { - t.Fatal("default registries are empty") - } - - for _, example := range maintainedExampleFiles(t) { - var stdout, stderr strings.Builder - code := RunWithOptions([]string{"config", "validate", "--config", example.path, "--pipeline", "dnd-session"}, &stdout, &stderr, Options{LookupEnv: emptyLookup}) - if code != 0 || stderr.Len() != 0 { - t.Fatalf("validate maintained %s config with defaults: code=%d stdout=%q stderr=%q", example.name, code, stdout.String(), stderr.String()) - } +func TestDefaultCLICompositionValidatesRepresentativeConfiguration(t *testing.T) { + var stdout, stderr strings.Builder + code := RunWithOptions([]string{ + "config", "validate", "--config", repositoryPath("examples", "dnd-spells.config.yml"), "--pipeline", "dnd-session", + }, &stdout, &stderr, Options{}) + if code != 0 || stderr.Len() != 0 { + t.Fatalf("validate representative config with default composition: code=%d stdout=%q stderr=%q", code, stdout.String(), stderr.String()) } } @@ -212,14 +187,6 @@ func TestProductionConfigValidationCoversModuleAndVariantFailures(t *testing.T) if code := RunWithOptions([]string{"config", "validate", "--config", validPath, "--pipeline", "dnd-session"}, &stdout, &stderr, options); code != 0 { t.Fatalf("valid production config: code=%d stdout=%q stderr=%q", code, stdout.String(), stderr.String()) } - for _, example := range maintainedExampleFiles(t) { - var exampleStdout, exampleStderr strings.Builder - code := RunWithOptions([]string{"config", "validate", "--config", example.path, "--pipeline", "dnd-session"}, &exampleStdout, &exampleStderr, options) - if code != 0 { - t.Fatalf("validate maintained %s config: code=%d stdout=%q stderr=%q", example.name, code, exampleStdout.String(), exampleStderr.String()) - } - } - tests := []struct { name string content string diff --git a/internal/cli/state_surfaces_test.go b/internal/cli/state_surfaces_test.go deleted file mode 100644 index 785edbe..0000000 --- a/internal/cli/state_surfaces_test.go +++ /dev/null @@ -1,96 +0,0 @@ -package cli - -import ( - "bytes" - "os" - "path/filepath" - "strings" - "testing" - "time" -) - -const stateSurfaceRunID = "run-1000000000-55555555555555555555555555555555" - -func stateSurfaceRunIDGenerator(time.Time) (string, error) { return stateSurfaceRunID, nil } - -func TestRunRejectsDebugDirectoryWithoutDebug(t *testing.T) { - var stdout, stderr bytes.Buffer - code := RunWithOptions([]string{"run", "example", "--input", "source.json", "--debug-dir", t.TempDir()}, &stdout, &stderr, Options{}) - if code != 2 || !strings.Contains(stderr.String(), "--debug-dir requires --debug") { - t.Fatalf("code=%d stderr=%q", code, stderr.String()) - } -} - -func TestRunDebugAllocatesBeforePipelineResolution(t *testing.T) { - root := t.TempDir() - configPath := writeV3Config(t, "") - var stdout, stderr bytes.Buffer - code := RunWithOptions([]string{"run", "missing", "--config", configPath, "--input", "source.json", "--debug", "--debug-dir", root, "--chunk_cache", "bypass"}, &stdout, &stderr, Options{LookupEnv: emptyLookup, RunIDGenerator: stateSurfaceRunIDGenerator}) - if code != 1 { - t.Fatalf("code=%d stderr=%q", code, stderr.String()) - } - entries, err := os.ReadDir(root) - if err != nil || len(entries) != 1 { - t.Fatalf("debug bundles: %v, %v", entries, err) - } - if entries[0].Name() != stateSurfaceRunID { - t.Fatalf("debug bundle name = %q, want %q", entries[0].Name(), stateSurfaceRunID) - } - bundle := filepath.Join(root, stateSurfaceRunID) - for _, name := range []string{"summary", "trace"} { - if info, err := os.Stat(filepath.Join(bundle, name)); err != nil || !info.IsDir() { - t.Fatalf("%s: %v", name, err) - } - } - if !strings.Contains(stderr.String(), "debug=") { - t.Fatalf("stderr does not include bundle path: %q", stderr.String()) - } -} - -func TestRunWithoutDebugDoesNotAllocateDebugRoot(t *testing.T) { - root := filepath.Join(t.TempDir(), "not-created") - configPath := writeV3Config(t, "") - var stdout, stderr bytes.Buffer - lookup := func(name string) (string, bool) { - if name == "NOTARIUS_DEBUG_DIR" { - return root, true - } - return "", false - } - code := RunWithOptions([]string{"run", "missing", "--config", configPath, "--input", "source.json", "--chunk_cache", "bypass"}, &stdout, &stderr, Options{LookupEnv: lookup, RunIDGenerator: stateSurfaceRunIDGenerator}) - if code != 1 { - t.Fatalf("code=%d stderr=%q", code, stderr.String()) - } - if _, err := os.Stat(root); !os.IsNotExist(err) { - t.Fatalf("debug root exists or unexpected error: %v", err) - } -} - -func TestConfigValidateUsesVersion3AndRemovedFieldsFail(t *testing.T) { - configPath := writeV3Config(t, "") - var stdout, stderr bytes.Buffer - if code := RunWithOptions([]string{"config", "validate", "--config", configPath}, &stdout, &stderr, Options{LookupEnv: emptyLookup}); code != 0 { - t.Fatalf("code=%d stderr=%q", code, stderr.String()) - } - legacy := filepath.Join(t.TempDir(), "legacy.yml") - if err := os.WriteFile(legacy, []byte("version: 3\nworkspace:\n directory: /tmp/old\n"), 0o600); err != nil { - t.Fatal(err) - } - stdout.Reset() - stderr.Reset() - if code := RunWithOptions([]string{"config", "validate", "--config", legacy}, &stdout, &stderr, Options{LookupEnv: emptyLookup}); code != 1 || !strings.Contains(stderr.String(), "field workspace not found") { - t.Fatalf("code=%d stderr=%q", code, stderr.String()) - } -} - -func writeV3Config(t *testing.T, extra string) string { - t.Helper() - path := filepath.Join(t.TempDir(), "config.yml") - data := "version: 3\noutput:\n directory: ./out\ncache:\n chunk_plans:\n mode: bypass\n checkpoints: {}\ndebug:\n directory: ./debug\n" + extra + "pipelines: {}\n" - if err := os.WriteFile(path, []byte(data), 0o600); err != nil { - t.Fatal(err) - } - return path -} - -func emptyLookup(string) (string, bool) { return "", false } diff --git a/internal/core/config/file_config_contract_test.go b/internal/core/config/file_config_contract_test.go index 21e30e7..93a84da 100644 --- a/internal/core/config/file_config_contract_test.go +++ b/internal/core/config/file_config_contract_test.go @@ -71,6 +71,11 @@ func TestFileConfigRejectsUnknownCurrentAndRemovedFields(t *testing.T) { yaml: "version: 3\nllm_profiles: {}\n", want: "field llm_profiles not found", }, + { + name: "version 2 migration", + yaml: "version: 2\nworkspace:\n directory: /tmp/old\n", + want: "version 2-to-3 migration", + }, { name: "pipeline field", yaml: "version: 3\npipelines:\n main:\n unknown: true\n", diff --git a/internal/core/config/v3_test.go b/internal/core/config/v3_test.go deleted file mode 100644 index 5795d7a..0000000 --- a/internal/core/config/v3_test.go +++ /dev/null @@ -1,124 +0,0 @@ -package config - -import ( - "encoding/json" - "path/filepath" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" -) - -func TestVersion3DefaultsAndValidation(t *testing.T) { - cfg := Default() - if cfg.Output.Directory != "./notarius-output" || cfg.Debug.Directory != "./notarius-debug" || cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheAuto { - t.Fatalf("unexpected defaults: %#v", cfg) - } - if err := cfg.Validate(); err != nil { - t.Fatalf("Validate() error = %v", err) - } -} - -func TestVersion3FileSchemaIsStrictAndRejectsVersion2BeforeDecode(t *testing.T) { - _, err := ParseFileConfigYAML([]byte("version: 2\nworkspace:\n directory: /tmp/old\n")) - if err == nil || !strings.Contains(err.Error(), "version 2-to-3 migration") { - t.Fatalf("version 2 error = %v", err) - } - _, err = ParseFileConfigYAML([]byte("version: 3\nworkspace:\n directory: /tmp/old\n")) - if err == nil || !strings.Contains(err.Error(), "field workspace not found") { - t.Fatalf("unknown field error = %v", err) - } - _, err = ParseFileConfigYAML([]byte("version: 4\n")) - if err == nil || !strings.Contains(err.Error(), "unsupported config version 4") { - t.Fatalf("version 4 error = %v", err) - } -} - -func TestStatePrecedenceAndInvalidSources(t *testing.T) { - file, err := ParseFileConfigYAML([]byte(`version: 3 -output: - directory: ./file-output -cache: - chunk_plans: - directory: ./plans - mode: refresh - checkpoints: - directory: ./checkpoints -debug: - directory: ./debug -`)) - if err != nil { - t.Fatal(err) - } - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - lookup := func(name string) (string, bool) { - values := map[string]string{ - "NOTARIUS_OUTPUT_DIR": "/env/output", "NOTARIUS_CACHE_CHUNK_PLANS_MODE": "auto", - "NOTARIUS_CACHE_CHUNK_PLANS_DIR": "/env/plans", "NOTARIUS_CACHE_CHECKPOINTS_DIR": "/env/checkpoints", "NOTARIUS_DEBUG_DIR": "/env/debug", - } - v, ok := values[name] - return v, ok - } - if err := cfg.ApplyEnvOverridesWithLookup(lookup); err != nil { - t.Fatal(err) - } - if cfg.Output.Directory != "/env/output" || cfg.Cache.ChunkPlans.Directory != "/env/plans" || cfg.Cache.Checkpoints.Directory != "/env/checkpoints" || cfg.Debug.Directory != "/env/debug" || cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheAuto { - t.Fatalf("unexpected environment precedence: %#v", cfg) - } - - bad := Default() - err = bad.ApplyEnvOverridesWithLookup(func(name string) (string, bool) { - if name == "NOTARIUS_DEBUG_DIR" { - return " ", true - } - return "", false - }) - if err == nil || !strings.Contains(err.Error(), "NOTARIUS_DEBUG_DIR") { - t.Fatalf("empty debug environment error = %v", err) - } - invalidFile, err := ParseFileConfigYAML([]byte("version: 3\noutput:\n directory: ' '\n")) - if err != nil { - t.Fatal(err) - } - bad = Default() - if err := bad.ApplyFileConfig(invalidFile); err == nil || !strings.Contains(err.Error(), "output.directory") { - t.Fatalf("invalid file error = %v", err) - } -} - -func TestRedactedSummaryContainsOnlyVersion3StateFields(t *testing.T) { - cfg := Default() - cfg.Pipelines["example"] = pipeline.PipelineProfile{Input: pipeline.ModuleBinding{Module: "input", Options: map[string]any{"api_key": "secret-value", "safe": "value"}}} - payload, err := json.Marshal(cfg.RedactedSummaryPayload()) - if err != nil { - t.Fatal(err) - } - text := string(payload) - for _, forbidden := range []string{"workspace", "diagnostics"} { - if strings.Contains(text, forbidden) { - t.Fatalf("payload contains %q: %s", forbidden, text) - } - } - if strings.Contains(text, "secret-value") || !strings.Contains(text, "[REDACTED]") { - t.Fatalf("payload did not redact sensitive option: %s", text) - } -} - -func TestCacheFamilyDefaultsAreIndependent(t *testing.T) { - base := filepath.Join(t.TempDir(), "cache") - resolver := func() (string, error) { return base, nil } - plans, err := DefaultChunkPlanRoot(resolver) - if err != nil { - t.Fatal(err) - } - checkpoints, err := DefaultCheckpointRoot(resolver) - if err != nil { - t.Fatal(err) - } - if plans == checkpoints || plans != filepath.Join(base, "notarius", "chunk-plans") || checkpoints != filepath.Join(base, "notarius", "checkpoints") { - t.Fatalf("roots = %q, %q", plans, checkpoints) - } -} diff --git a/internal/framework/pipeline/default_modules_test.go b/internal/framework/pipeline/default_modules_test.go deleted file mode 100644 index 6e01a00..0000000 --- a/internal/framework/pipeline/default_modules_test.go +++ /dev/null @@ -1,155 +0,0 @@ -package pipeline_test - -import ( - "context" - "encoding/json" - "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" - "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/chunk/units" - "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/merge/appendorder" - "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/normalize/noop" - jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/output/json" -) - -func TestPipelineConfigResolvesWithProductionDefaultsRegistered(t *testing.T) { - cfg := config.Default() - cfg.Pipelines = map[string]pipeline.PipelineProfile{ - "defaults": { - Input: pipeline.Binding("input"), - Artifacts: map[string]pipeline.ArtifactLaneProfile{ - "events": {Extract: pipeline.Binding("extract")}, - }, - }, - } - - resolved, err := cfg.Resolve(config.ResolveInput{ - PipelineID: "defaults", - Catalog: defaultModuleCatalog(t), - }) - if err != nil { - t.Fatalf("Resolve() error = %v, want nil", err) - } - - pipeline := resolved.ResolvedPipeline - if pipeline.Chunk.Module != units.Key { - t.Fatalf("Chunk.Module = %q, want %q", pipeline.Chunk.Module, units.Key) - } - if pipeline.Output.Module != jsonoutput.Key { - t.Fatalf("Output.Module = %q, want %q", pipeline.Output.Module, jsonoutput.Key) - } - lane := pipeline.ArtifactLanes[0] - if lane.Merge.Module != appendorder.Key { - t.Fatalf("Merge.Module = %q, want %q", lane.Merge.Module, appendorder.Key) - } - if lane.Normalize.Module != noop.Key { - t.Fatalf("Normalize.Module = %q, want %q", lane.Normalize.Module, noop.Key) - } -} - -func defaultModuleCatalog(t *testing.T) pipeline.ModuleCatalog { - t.Helper() - - inputs := pipeline.NewInputAdapterRegistry() - chunkers := pipeline.NewChunkerRegistry() - extractors := pipeline.NewExtractorRegistry() - mergers := pipeline.NewMergerRegistry() - normalizers := pipeline.NewNormalizerRegistry() - outputs := pipeline.NewOutputEncoderRegistry() - - if err := inputs.RegisterWithSpec(pipeline.ModuleSpec{ - Key: "input", - Stage: pipeline.StageInput, - Provides: []string{"source"}, - }, func() (contracts.InputAdapter, error) { - return defaultInput{}, nil - }); err != nil { - t.Fatalf("register input: %v", err) - } - if err := units.Register(chunkers); err != nil { - t.Fatalf("register generic chunker: %v", err) - } - if err := pipeline.RegisterExtractor[defaultArtifact](extractors, pipeline.ModuleSpec{ - Key: "extract", - Stage: pipeline.StageExtract, - ArtifactKind: defaultArtifactKind, - Requires: []string{"chunks"}, - Provides: []string{"records"}, - }, func() (contracts.Extractor[defaultArtifact], error) { - return defaultExtractor{}, nil - }); err != nil { - t.Fatalf("register extractor: %v", err) - } - if err := appendorder.RegisterTyped(mergers, defaultArtifactKind, func(values []defaultArtifact) (defaultArtifact, error) { - if len(values) == 0 { - return defaultArtifact{}, nil - } - return values[0], nil - }); err != nil { - t.Fatalf("register appendorder merger: %v", err) - } - if err := noop.RegisterTyped[defaultArtifact](normalizers, defaultArtifactKind); err != nil { - t.Fatalf("register noop normalizer: %v", err) - } - if err := jsonoutput.Register(outputs); err != nil { - t.Fatalf("register json output: %v", err) - } - - codecs := pipeline.NewArtifactCodecRegistry() - if err := pipeline.RegisterArtifactCodec(codecs, defaultArtifactCodec{}); 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, - } -} - -type defaultInput struct{} - -func (defaultInput) Key() string { return "input" } - -func (defaultInput) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) { - return nil, nil -} - -type defaultExtractor struct{} - -func (defaultExtractor) Key() string { return "extract" } - -func (defaultExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil } - -func (defaultExtractor) Extract(ctx context.Context, req contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[defaultArtifact], error) { - return contracts.TypedExtractionResult[defaultArtifact]{}, nil -} - -const defaultArtifactKind contracts.ArtifactKind = "test/default" - -type defaultArtifact struct { - Value string `json:"value"` -} -type defaultArtifactCodec struct{} - -func (defaultArtifactCodec) Kind() contracts.ArtifactKind { return defaultArtifactKind } -func (defaultArtifactCodec) Schema() contracts.ArtifactSchema { - return contracts.ArtifactSchema{ID: "urn:notarius:test:default", Name: "default", Version: "1", JSONSchema: []byte(`{"type":"object"}`)} -} -func (defaultArtifactCodec) MediaType() string { return "application/json" } -func (defaultArtifactCodec) EncodeCandidate(value defaultArtifact) ([]byte, error) { - return json.Marshal(value) -} -func (defaultArtifactCodec) Encode(value defaultArtifact) ([]byte, error) { return json.Marshal(value) } -func (defaultArtifactCodec) Decode(content []byte) (defaultArtifact, error) { - var value defaultArtifact - err := json.Unmarshal(content, &value) - return value, err -} diff --git a/internal/modules/integration/dnd_spells_config_test.go b/internal/modules/integration/dnd_spells_config_test.go index 65a3225..4de698e 100644 --- a/internal/modules/integration/dnd_spells_config_test.go +++ b/internal/modules/integration/dnd_spells_config_test.go @@ -18,109 +18,102 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/modules/seriatim/input/transcript" ) -func TestPipelineConfigLoadsAndResolvesWithDNDSpellsExtractor(t *testing.T) { - 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) +func TestDNDSpellCapabilityFailures(t *testing.T) { + tests := []struct { + name string + mutate func(pipeline.ModuleSpec, pipeline.ModuleSpec) (pipeline.ModuleSpec, pipeline.ModuleSpec) + wantModule string + wantCap string + }{ + { + name: "spell extractor requires transcript source", + mutate: func(input, extractor pipeline.ModuleSpec) (pipeline.ModuleSpec, pipeline.ModuleSpec) { + input.Provides = withoutCapability(input.Provides, "source.transcript") + return input, extractor + }, + wantModule: spells.Key, + wantCap: "source.transcript", + }, + { + name: "append-order merger requires spell casts", + mutate: func(input, extractor pipeline.ModuleSpec) (pipeline.ModuleSpec, pipeline.ModuleSpec) { + extractor.Provides = withoutCapability(extractor.Provides, "dnd.spell_casts") + return input, extractor + }, + wantModule: pipeline.DefaultMergeModule, + wantCap: "dnd.spell_casts", + }, } - cfg := config.Default() - if err := cfg.ApplyFileConfig(fileCfg); err != nil { - t.Fatalf("ApplyFileConfig() error = %v, want nil", err) - } - - resolved, err := cfg.Resolve(config.ResolveInput{ - PipelineID: "dnd-spells-fixture", - Catalog: dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{}), - }) - if err != nil { - t.Fatalf("Resolve() error = %v, want nil", err) - } - - if len(resolved.ResolvedPipeline.ArtifactLanes) != 1 { - t.Fatalf("len(ArtifactLanes) = %d, want 1", len(resolved.ResolvedPipeline.ArtifactLanes)) - } - lane := resolved.ResolvedPipeline.ArtifactLanes[0] - if lane.ID != "spells" { - t.Fatalf("lane ID = %q, want spells", lane.ID) - } - if lane.Extract.Module != spells.Key { - t.Fatalf("extract module = %q, want %q", lane.Extract.Module, spells.Key) - } - if resolved.ResolvedPipeline.Digest == "" { - t.Fatal("resolved digest is empty") - } - - again, err := cfg.Resolve(config.ResolveInput{ - PipelineID: "dnd-spells-fixture", - Catalog: dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{}), - }) - 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) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + inputSpec, extractorSpec := tt.mutate(transcript.ModuleSpec(), spells.ModuleSpec()) + _, err := pipeline.ResolvePipeline(dndCapabilityProfile(), pipeline.ResolveOptions{}, dndCapabilityCatalog(t, inputSpec, extractorSpec)) + if err == nil || !strings.Contains(err.Error(), "missing capability") || !strings.Contains(err.Error(), tt.wantCap) || !strings.Contains(err.Error(), tt.wantModule) { + t.Fatalf("ResolvePipeline() error = %v, want %s missing %s capability", err, tt.wantModule, tt.wantCap) + } + }) } } -func TestPipelineConfigRejectsMissingTranscriptCapabilityForDNDSpells(t *testing.T) { - inputSpec := transcript.ModuleSpec() - inputSpec.Provides = withoutCapability(inputSpec.Provides, "source.transcript") - chunkSpec := dndSpellsChunkerSpec() - chunkSpec.Requires = nil - - _, err := loadDNDSpellsPipelineConfig(t).Resolve(config.ResolveInput{ - PipelineID: "dnd-spells-fixture", - Catalog: dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{ - input: inputSpec, - chunk: chunkSpec, - }), - }) - if err == nil { - t.Fatal("Resolve() error = nil, want missing capability error") - } - if !strings.Contains(err.Error(), "missing capability") || - !strings.Contains(err.Error(), "source.transcript") || - !strings.Contains(err.Error(), spells.Key) { - t.Fatalf("Resolve() error = %q, want dnd/spells missing source.transcript capability", err.Error()) +func dndCapabilityProfile() pipeline.PipelineProfile { + return pipeline.PipelineProfile{ + ID: "dnd-capability", + Input: pipeline.Binding(transcript.Key), + Chunk: pipeline.Binding("fake/chunk"), + Artifacts: map[string]pipeline.ArtifactLaneProfile{ + "spells": {Extract: pipeline.Binding(spells.Key)}, + }, } } -func TestPipelineConfigRejectsMissingSpellCastsCapabilityForAppendOrder(t *testing.T) { - extractorSpec := spells.ModuleSpec() - extractorSpec.Provides = withoutCapability(extractorSpec.Provides, "dnd.spell_casts") +func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.ModuleSpec) pipeline.ModuleCatalog { + t.Helper() + inputs := pipeline.NewInputAdapterRegistry() + if err := inputs.RegisterWithSpec(inputSpec, func() (contracts.InputAdapter, error) { return transcript.New(), nil }); err != nil { + t.Fatalf("register capability input: %v", err) + } - _, err := loadDNDSpellsPipelineConfig(t).Resolve(config.ResolveInput{ - PipelineID: "dnd-spells-fixture", - Catalog: dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{ - extractor: extractorSpec, - }), - }) - if err == nil { - t.Fatal("Resolve() error = nil, want missing capability error") + chunkers := pipeline.NewChunkerRegistry() + if err := chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "fake/chunk", Stage: pipeline.StageChunk, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) { return dndSpellsChunker{}, nil }); err != nil { + t.Fatalf("register capability chunker: %v", err) } - if !strings.Contains(err.Error(), "missing capability") || - !strings.Contains(err.Error(), "dnd.spell_casts") || - !strings.Contains(err.Error(), pipeline.DefaultMergeModule) { - t.Fatalf("Resolve() error = %q, want appendorder missing dnd.spell_casts capability", err.Error()) - } -} -func TestPipelineConfigRejectsUnknownLaneSelection(t *testing.T) { - _, err := loadDNDSpellsPipelineConfig(t).Resolve(config.ResolveInput{ - PipelineID: "dnd-spells-fixture", - Only: []string{"missing"}, - Catalog: dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{}), - }) - if err == nil { - t.Fatal("Resolve() error = nil, want unknown lane error") + extractors := pipeline.NewExtractorRegistry() + extractorSpec.ArtifactKind = dnd.SpellListKind + if err := pipeline.RegisterExtractor[dnd.SpellList](extractors, extractorSpec, func() (contracts.Extractor[dnd.SpellList], error) { + return configExtractor{key: extractorSpec.Key}, nil + }); err != nil { + t.Fatalf("register capability extractor: %v", err) } - if !strings.Contains(err.Error(), "selected artifact lane") || !strings.Contains(err.Error(), "missing") { - t.Fatalf("Resolve() error = %q, want unknown lane context", err.Error()) + + codecs := pipeline.NewArtifactCodecRegistry() + if err := pipeline.RegisterArtifactCodec(codecs, spellcodec.New()); err != nil { + t.Fatalf("register capability codec: %v", err) + } + + mergers := pipeline.NewMergerRegistry() + if err := pipeline.RegisterMerger[dnd.SpellList](mergers, pipeline.ModuleSpec{ + Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: dnd.SpellListKind, Requires: []string{"dnd.spell_casts"}, + }, func() (contracts.Merger[dnd.SpellList], error) { return appendorder.NewTyped(appendSpellLists) }); err != nil { + t.Fatalf("register capability merger: %v", err) + } + + normalizers := pipeline.NewNormalizerRegistry() + if err := pipeline.RegisterNormalizer[dnd.SpellList](normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: dnd.SpellListKind}, func() (contracts.Normalizer[dnd.SpellList], error) { + return noop.NewTyped[dnd.SpellList](), nil + }); err != nil { + t.Fatalf("register capability normalizer: %v", err) + } + + outputs := pipeline.NewOutputEncoderRegistry() + if err := outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput}, func() (contracts.OutputEncoder, error) { return dndSpellsOutput{}, nil }); err != nil { + t.Fatalf("register capability output: %v", err) + } + + return pipeline.ModuleCatalog{ + Inputs: inputs, Chunkers: chunkers, ArtifactCodecs: codecs, Extractors: extractors, + Mergers: mergers, Normalizers: normalizers, ValidatorChains: pipeline.NewValidatorChainRegistry(), Outputs: outputs, } } diff --git a/internal/modules/seriatim/input/transcript/config_test.go b/internal/modules/seriatim/input/transcript/config_test.go deleted file mode 100644 index 458b492..0000000 --- a/internal/modules/seriatim/input/transcript/config_test.go +++ /dev/null @@ -1,306 +0,0 @@ -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 -} diff --git a/internal/modules/seriatim/input/transcript/runner_helpers_test.go b/internal/modules/seriatim/input/transcript/runner_helpers_test.go new file mode 100644 index 0000000..6c60e3f --- /dev/null +++ b/internal/modules/seriatim/input/transcript/runner_helpers_test.go @@ -0,0 +1,166 @@ +package transcript + +import ( + "context" + "encoding/json" + "reflect" + "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 loadPipelineConfig(t *testing.T) config.Config { + t.Helper() + + cfg := config.Default() + cfg.Pipelines["seriatim-fixture"] = pipeline.PipelineProfile{ + ID: "seriatim-fixture", + Input: pipeline.Binding(Key), + Chunk: pipeline.Binding("fake/chunk"), + Artifacts: map[string]pipeline.ArtifactLaneProfile{ + "events": {Extract: pipeline.Binding("fake/extract")}, + }, + } + 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(context.Context, 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(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[seriatimArtifact], error) { + return contracts.TypedExtractionResult[seriatimArtifact]{}, nil +} + +type fakeMerger struct{} + +func (fakeMerger) Key() string { return pipeline.DefaultMergeModule } +func (fakeMerger) Merge(_ 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(_ 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(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) { + return contracts.OutputResult{}, nil +} + +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 +} diff --git a/internal/modules/seriatim/input/transcript/testdata/pipeline.yml b/internal/modules/seriatim/input/transcript/testdata/pipeline.yml deleted file mode 100644 index 612dc5c..0000000 --- a/internal/modules/seriatim/input/transcript/testdata/pipeline.yml +++ /dev/null @@ -1,19 +0,0 @@ -version: 3 -output: - directory: ./notarius-output -cache: - chunk_plans: - mode: bypass - checkpoints: {} -debug: - directory: ./notarius-debug -pipelines: - seriatim-fixture: - input: seriatim - chunk: fake/chunk - artifacts: - events: - extract: fake/extract - merge: appendorder - normalize: noop - output: json