From 0ad96618fcce20380a1d5802cb95c886fbc02b9b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 4 Jul 2026 00:55:18 +0000 Subject: [PATCH] Add production default pipeline modules --- .../pipeline/default_modules_test.go | 124 +++++++ internal/framework/pipeline/generic_stages.go | 70 ---- .../framework/pipeline/generic_stages_test.go | 221 ------------ .../pipeline/walking_skeleton_test.go | 28 +- internal/modules/chunk/generic/chunker.go | 241 +++++++++++++ .../modules/chunk/generic/chunker_test.go | 213 ++++++++++++ .../modules/extract/dnd/spells/config_test.go | 6 +- .../modules/input/seriatim/config_test.go | 6 +- .../modules/input/seriatim/runner_test.go | 6 +- internal/modules/merge/appendorder/merger.go | 97 ++++++ .../modules/merge/appendorder/merger_test.go | 147 ++++++++ internal/modules/normalize/noop/normalizer.go | 89 +++++ .../modules/normalize/noop/normalizer_test.go | 133 ++++++++ internal/modules/output/json/encoder.go | 253 ++++++++++++++ internal/modules/output/json/encoder_test.go | 316 ++++++++++++++++++ 15 files changed, 1651 insertions(+), 299 deletions(-) create mode 100644 internal/framework/pipeline/default_modules_test.go delete mode 100644 internal/framework/pipeline/generic_stages.go delete mode 100644 internal/framework/pipeline/generic_stages_test.go create mode 100644 internal/modules/chunk/generic/chunker.go create mode 100644 internal/modules/chunk/generic/chunker_test.go create mode 100644 internal/modules/merge/appendorder/merger.go create mode 100644 internal/modules/merge/appendorder/merger_test.go create mode 100644 internal/modules/normalize/noop/normalizer.go create mode 100644 internal/modules/normalize/noop/normalizer_test.go create mode 100644 internal/modules/output/json/encoder.go create mode 100644 internal/modules/output/json/encoder_test.go diff --git a/internal/framework/pipeline/default_modules_test.go b/internal/framework/pipeline/default_modules_test.go new file mode 100644 index 0000000..603d621 --- /dev/null +++ b/internal/framework/pipeline/default_modules_test.go @@ -0,0 +1,124 @@ +package pipeline_test + +import ( + "context" + "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/chunk/generic" + "gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder" + "gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop" + jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/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 != generic.Key { + t.Fatalf("Chunk.Module = %q, want %q", pipeline.Chunk.Module, generic.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 := generic.Register(chunkers); err != nil { + t.Fatalf("register generic chunker: %v", err) + } + if err := extractors.RegisterWithSpec(pipeline.ModuleSpec{ + Key: "extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"records"}, + }, func() (contracts.Extractor, error) { + return defaultExtractor{}, nil + }); err != nil { + t.Fatalf("register extractor: %v", err) + } + if err := appendorder.Register(mergers); err != nil { + t.Fatalf("register appendorder merger: %v", err) + } + if err := noop.Register(normalizers); err != nil { + t.Fatalf("register noop normalizer: %v", err) + } + if err := jsonoutput.Register(outputs); err != nil { + t.Fatalf("register json output: %v", err) + } + + return pipeline.ModuleCatalog{ + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + 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) ArtifactType() string { return "record" } + +func (defaultExtractor) SchemaVersion() string { return "v1" } + +func (defaultExtractor) Validators() []contracts.Validator { return nil } + +func (defaultExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) { + return contracts.ExtractionResult{}, nil +} diff --git a/internal/framework/pipeline/generic_stages.go b/internal/framework/pipeline/generic_stages.go deleted file mode 100644 index 340d8db..0000000 --- a/internal/framework/pipeline/generic_stages.go +++ /dev/null @@ -1,70 +0,0 @@ -package pipeline - -import ( - "context" - "encoding/json" - - "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" - "gitea.maximumdirect.net/eric/notarius/internal/core/source" - "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" -) - -type AppendOrderMerger struct{} - -func (m AppendOrderMerger) Key() string { - return DefaultMergeModule -} - -func (m AppendOrderMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) { - var candidates []artifacts.ArtifactCandidate - for _, chunkArtifacts := range req.ChunkArtifacts { - candidates = append(candidates, copyArtifactCandidates(chunkArtifacts.Candidates)...) - } - return contracts.MergeResult{Candidates: candidates}, nil -} - -type NoopNormalizer struct{} - -func (n NoopNormalizer) Key() string { - return DefaultNormalizeModule -} - -func (n NoopNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) { - return contracts.NormalizeResult{Candidates: copyArtifactCandidates(req.Candidates)}, nil -} - -func copyArtifactCandidates(candidates []artifacts.ArtifactCandidate) []artifacts.ArtifactCandidate { - if len(candidates) == 0 { - return nil - } - - copied := make([]artifacts.ArtifactCandidate, 0, len(candidates)) - for _, candidate := range candidates { - copied = append(copied, copyArtifactCandidate(candidate)) - } - return copied -} - -func copyArtifactCandidate(candidate artifacts.ArtifactCandidate) artifacts.ArtifactCandidate { - return artifacts.ArtifactCandidate{ - Index: candidate.Index, - ExtractorKey: candidate.ExtractorKey, - ArtifactType: candidate.ArtifactType, - SchemaVersion: candidate.SchemaVersion, - Payload: append(json.RawMessage(nil), candidate.Payload...), - SourceRefs: append([]source.SourceRef(nil), candidate.SourceRefs...), - Metadata: copyArtifactMetadata(candidate.Metadata), - } -} - -func copyArtifactMetadata(metadata map[string]any) map[string]any { - if len(metadata) == 0 { - return nil - } - - copied := make(map[string]any, len(metadata)) - for key, value := range metadata { - copied[key] = value - } - return copied -} diff --git a/internal/framework/pipeline/generic_stages_test.go b/internal/framework/pipeline/generic_stages_test.go deleted file mode 100644 index 400ffde..0000000 --- a/internal/framework/pipeline/generic_stages_test.go +++ /dev/null @@ -1,221 +0,0 @@ -package pipeline - -import ( - "context" - "encoding/json" - "reflect" - "testing" - - "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" - "gitea.maximumdirect.net/eric/notarius/internal/core/source" - "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" -) - -func TestGenericMergeAndNormalizeKeys(t *testing.T) { - merger := AppendOrderMerger{} - normalizer := NoopNormalizer{} - - if merger.Key() != DefaultMergeModule { - t.Fatalf("AppendOrderMerger.Key() = %q, want %q", merger.Key(), DefaultMergeModule) - } - if normalizer.Key() != DefaultNormalizeModule { - t.Fatalf("NoopNormalizer.Key() = %q, want %q", normalizer.Key(), DefaultNormalizeModule) - } -} - -func TestAppendOrderMergerConcatenatesByChunkAndCandidateOrder(t *testing.T) { - merger := AppendOrderMerger{} - chunks := []contracts.ChunkArtifacts{ - { - Chunk: sourceChunk(0), - Candidates: []artifacts.ArtifactCandidate{ - candidate(2, "first-b"), - candidate(1, "first-a"), - }, - }, - { - Chunk: sourceChunk(1), - Candidates: []artifacts.ArtifactCandidate{ - candidate(4, "second-b"), - candidate(3, "second-a"), - }, - }, - } - - result, err := merger.Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: chunks}) - if err != nil { - t.Fatalf("Merge() error = %v, want nil", err) - } - if len(result.Warnings) != 0 { - t.Fatalf("Warnings = %#v, want none", result.Warnings) - } - - got := candidateNames(result.Candidates) - want := []string{"first-b", "first-a", "second-b", "second-a"} - if !reflect.DeepEqual(got, want) { - t.Fatalf("candidate order = %#v, want %#v", got, want) - } -} - -func TestAppendOrderMergerReturnsMutationSafeCandidates(t *testing.T) { - merger := AppendOrderMerger{} - input := []contracts.ChunkArtifacts{ - { - Chunk: sourceChunk(0), - Candidates: []artifacts.ArtifactCandidate{ - candidate(1, "original"), - }, - }, - } - - result, err := merger.Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: input}) - if err != nil { - t.Fatalf("Merge() error = %v, want nil", err) - } - if len(result.Candidates) != 1 { - t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates)) - } - - input[0].Candidates[0].Index = 99 - input[0].Candidates[0].Payload[0] = '[' - input[0].Candidates[0].SourceRefs[0].StartUnitID = "changed" - input[0].Candidates[0].Metadata["name"] = "changed" - - got := result.Candidates[0] - if got.Index != 1 { - t.Fatalf("Index = %d, want 1", got.Index) - } - if string(got.Payload) != `{"name":"original"}` { - t.Fatalf("Payload = %s, want original payload", got.Payload) - } - if got.SourceRefs[0].StartUnitID != "u1" { - t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs) - } - if got.Metadata["name"] != "original" { - t.Fatalf("Metadata = %#v, want original metadata", got.Metadata) - } -} - -func TestNoopNormalizerPreservesOrderAndValues(t *testing.T) { - normalizer := NoopNormalizer{} - input := []artifacts.ArtifactCandidate{ - candidate(3, "third"), - candidate(1, "first"), - candidate(2, "second"), - } - - result, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input}) - if err != nil { - t.Fatalf("Normalize() error = %v, want nil", err) - } - if len(result.Warnings) != 0 { - t.Fatalf("Warnings = %#v, want none", result.Warnings) - } - - got := candidateNames(result.Candidates) - want := []string{"third", "first", "second"} - if !reflect.DeepEqual(got, want) { - t.Fatalf("candidate order = %#v, want %#v", got, want) - } - - if !reflect.DeepEqual(result.Candidates[0].SourceRefs, input[0].SourceRefs) { - t.Fatalf("SourceRefs = %#v, want %#v", result.Candidates[0].SourceRefs, input[0].SourceRefs) - } - if !reflect.DeepEqual(result.Candidates[0].Metadata, input[0].Metadata) { - t.Fatalf("Metadata = %#v, want %#v", result.Candidates[0].Metadata, input[0].Metadata) - } -} - -func TestNoopNormalizerReturnsMutationSafeCandidates(t *testing.T) { - normalizer := NoopNormalizer{} - input := []artifacts.ArtifactCandidate{candidate(1, "original")} - - result, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input}) - if err != nil { - t.Fatalf("Normalize() error = %v, want nil", err) - } - if len(result.Candidates) != 1 { - t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates)) - } - - input[0].Index = 99 - input[0].Payload[0] = '[' - input[0].SourceRefs[0].EndUnitID = "changed" - input[0].Metadata["name"] = "changed" - - got := result.Candidates[0] - if got.Index != 1 { - t.Fatalf("Index = %d, want 1", got.Index) - } - if string(got.Payload) != `{"name":"original"}` { - t.Fatalf("Payload = %s, want original payload", got.Payload) - } - if got.SourceRefs[0].EndUnitID != "u1" { - t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs) - } - if got.Metadata["name"] != "original" { - t.Fatalf("Metadata = %#v, want original metadata", got.Metadata) - } -} - -func TestGenericMergeAndNormalizeHandleEmptyInput(t *testing.T) { - merger := AppendOrderMerger{} - normalizer := NoopNormalizer{} - - mergeResult, err := merger.Merge(context.Background(), contracts.MergeRequest{}) - if err != nil { - t.Fatalf("Merge() error = %v, want nil", err) - } - if len(mergeResult.Candidates) != 0 { - t.Fatalf("len(mergeResult.Candidates) = %d, want 0", len(mergeResult.Candidates)) - } - if len(mergeResult.Warnings) != 0 { - t.Fatalf("merge warnings = %#v, want none", mergeResult.Warnings) - } - - normalizeResult, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{}) - if err != nil { - t.Fatalf("Normalize() error = %v, want nil", err) - } - if len(normalizeResult.Candidates) != 0 { - t.Fatalf("len(normalizeResult.Candidates) = %d, want 0", len(normalizeResult.Candidates)) - } - if len(normalizeResult.Warnings) != 0 { - t.Fatalf("normalize warnings = %#v, want none", normalizeResult.Warnings) - } -} - -func candidate(index int, name string) artifacts.ArtifactCandidate { - return artifacts.ArtifactCandidate{ - Index: index, - ExtractorKey: "generic-extractor", - ArtifactType: "generic-artifact", - SchemaVersion: "v1", - Payload: json.RawMessage(`{"name":"` + name + `"}`), - SourceRefs: []source.SourceRef{ - {SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"}, - }, - Metadata: map[string]any{ - "name": name, - }, - } -} - -func candidateNames(candidates []artifacts.ArtifactCandidate) []string { - names := make([]string, 0, len(candidates)) - for _, candidate := range candidates { - names = append(names, candidate.Metadata["name"].(string)) - } - return names -} - -func sourceChunk(index int) contracts.SourceChunk { - return contracts.SourceChunk{ - ID: "chunk", - SourceID: "source-1", - Index: index, - Units: []source.SourceUnit{ - {ID: "u1", Kind: "unit", Text: "Source unit."}, - }, - } -} diff --git a/internal/framework/pipeline/walking_skeleton_test.go b/internal/framework/pipeline/walking_skeleton_test.go index bac1954..fe9af0b 100644 --- a/internal/framework/pipeline/walking_skeleton_test.go +++ b/internal/framework/pipeline/walking_skeleton_test.go @@ -135,7 +135,7 @@ func walkingSkeletonCatalog(t *testing.T) ModuleCatalog { Stage: StageMerge, Requires: []string{"fake_artifacts"}, }, func() (contracts.Merger, error) { - return AppendOrderMerger{}, nil + return walkingSkeletonMerger{}, nil }); err != nil { t.Fatalf("register append-order merger: %v", err) } @@ -143,7 +143,7 @@ func walkingSkeletonCatalog(t *testing.T) ModuleCatalog { Key: DefaultNormalizeModule, Stage: StageNormalize, }, func() (contracts.Normalizer, error) { - return NoopNormalizer{}, nil + return walkingSkeletonNormalizer{}, nil }); err != nil { t.Fatalf("register no-op normalizer: %v", err) } @@ -309,6 +309,30 @@ func (client *walkingSkeletonLLMClient) CompleteStructured(ctx context.Context, }, nil } +type walkingSkeletonMerger struct{} + +func (merger walkingSkeletonMerger) Key() string { + return DefaultMergeModule +} + +func (merger walkingSkeletonMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) { + var candidates []artifacts.ArtifactCandidate + for _, chunkArtifacts := range req.ChunkArtifacts { + candidates = append(candidates, chunkArtifacts.Candidates...) + } + return contracts.MergeResult{Candidates: candidates}, nil +} + +type walkingSkeletonNormalizer struct{} + +func (normalizer walkingSkeletonNormalizer) Key() string { + return DefaultNormalizeModule +} + +func (normalizer walkingSkeletonNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) { + return contracts.NormalizeResult{Candidates: req.Candidates}, nil +} + type walkingSkeletonOutput struct{} func (output walkingSkeletonOutput) Key() string { diff --git a/internal/modules/chunk/generic/chunker.go b/internal/modules/chunk/generic/chunker.go new file mode 100644 index 0000000..36e1036 --- /dev/null +++ b/internal/modules/chunk/generic/chunker.go @@ -0,0 +1,241 @@ +package generic + +import ( + "context" + "encoding/json" + "fmt" + "math" + "strconv" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const Key = "generic" + +const ( + defaultMaxUnits = 50 + defaultOverlapUnits = 0 +) + +var _ contracts.Chunker = (*Chunker)(nil) + +type Chunker struct{} + +func New() *Chunker { + return &Chunker{} +} + +func (c *Chunker) Key() string { + return Key +} + +func (c *Chunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) { + if c == nil { + return contracts.ChunkResult{}, chunkerErrorf("chunker must not be nil") + } + if ctx == nil { + return contracts.ChunkResult{}, chunkerErrorf("context must not be nil") + } + if err := ctx.Err(); err != nil { + return contracts.ChunkResult{}, chunkerErrorf("context error before chunking: %w", err) + } + if req.Source == nil { + return contracts.ChunkResult{}, chunkerErrorf("source must not be nil") + } + if len(req.Source.Units) == 0 { + return contracts.ChunkResult{}, chunkerErrorf("source units must not be empty") + } + if err := source.ValidateDocument(req.Source); err != nil { + return contracts.ChunkResult{}, chunkerErrorf("validate source document: %w", err) + } + + opts, err := chunkOptionsFrom(req.Options) + if err != nil { + return contracts.ChunkResult{}, err + } + + step := opts.maxUnits - opts.overlapUnits + chunks := make([]contracts.SourceChunk, 0, (len(req.Source.Units)+step-1)/step) + for start := 0; start < len(req.Source.Units); start += step { + end := start + opts.maxUnits + if end > len(req.Source.Units) { + end = len(req.Source.Units) + } + units := cloneUnits(req.Source.Units[start:end]) + chunks = append(chunks, contracts.SourceChunk{ + ID: fmt.Sprintf("chunk-%06d", len(chunks)+1), + SourceID: req.Source.ID, + Index: len(chunks), + Units: units, + Metadata: map[string]any{ + "start_unit_id": units[0].ID, + "end_unit_id": units[len(units)-1].ID, + "unit_count": len(units), + }, + }) + if end == len(req.Source.Units) { + break + } + } + + return contracts.ChunkResult{Chunks: chunks}, nil +} + +func ModuleSpec() pipeline.ModuleSpec { + return pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageChunk, + Provides: []string{"chunks"}, + } +} + +func Register(registry *pipeline.ChunkerRegistry) error { + return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.Chunker, error) { + return New(), nil + }) +} + +type chunkOptions struct { + maxUnits int + overlapUnits int +} + +func chunkOptionsFrom(options map[string]any) (chunkOptions, error) { + opts := chunkOptions{ + maxUnits: defaultMaxUnits, + overlapUnits: defaultOverlapUnits, + } + var err error + if value, ok := options["max_units"]; ok { + opts.maxUnits, err = positiveIntOption("max_units", value) + if err != nil { + return chunkOptions{}, err + } + } + if value, ok := options["overlap_units"]; ok { + opts.overlapUnits, err = nonNegativeIntOption("overlap_units", value) + if err != nil { + return chunkOptions{}, err + } + } + if opts.overlapUnits >= opts.maxUnits { + return chunkOptions{}, chunkerErrorf("overlap_units must be less than max_units") + } + return opts, nil +} + +func positiveIntOption(name string, value any) (int, error) { + got, err := intOption(name, value) + if err != nil { + return 0, err + } + if got <= 0 { + return 0, chunkerErrorf("%s must be positive", name) + } + return got, nil +} + +func nonNegativeIntOption(name string, value any) (int, error) { + got, err := intOption(name, value) + if err != nil { + return 0, err + } + if got < 0 { + return 0, chunkerErrorf("%s must be non-negative", name) + } + return got, nil +} + +func intOption(name string, value any) (int, error) { + switch typed := value.(type) { + case int: + return typed, nil + case int8: + return int(typed), nil + case int16: + return int(typed), nil + case int32: + return int(typed), nil + case int64: + if typed > maxInt() || typed < minInt() { + return 0, chunkerErrorf("%s is outside supported integer range", name) + } + return int(typed), nil + case uint: + if uint64(typed) > uint64(maxInt()) { + return 0, chunkerErrorf("%s is outside supported integer range", name) + } + return int(typed), nil + case uint8: + return int(typed), nil + case uint16: + return int(typed), nil + case uint32: + if uint64(typed) > uint64(maxInt()) { + return 0, chunkerErrorf("%s is outside supported integer range", name) + } + return int(typed), nil + case uint64: + if typed > uint64(maxInt()) { + return 0, chunkerErrorf("%s is outside supported integer range", name) + } + return int(typed), nil + case float64: + if typed != math.Trunc(typed) { + return 0, chunkerErrorf("%s must be an integer", name) + } + if typed > float64(maxInt()) || typed < float64(minInt()) { + return 0, chunkerErrorf("%s is outside supported integer range", name) + } + return int(typed), nil + case json.Number: + parsed, err := typed.Int64() + if err != nil { + return 0, chunkerErrorf("%s must be an integer", name) + } + if parsed > maxInt() || parsed < minInt() { + return 0, chunkerErrorf("%s is outside supported integer range", name) + } + return int(parsed), nil + default: + return 0, chunkerErrorf("%s must be an integer", name) + } +} + +func maxInt() int64 { + return int64(1<<(strconv.IntSize-1) - 1) +} + +func minInt() int64 { + return -maxInt() - 1 +} + +func cloneUnits(units []source.SourceUnit) []source.SourceUnit { + out := make([]source.SourceUnit, 0, len(units)) + for _, unit := range units { + out = append(out, source.SourceUnit{ + ID: unit.ID, + Kind: unit.Kind, + Text: unit.Text, + Metadata: cloneMetadata(unit.Metadata), + }) + } + return out +} + +func cloneMetadata(metadata map[string]any) map[string]any { + if len(metadata) == 0 { + return nil + } + out := make(map[string]any, len(metadata)) + for key, value := range metadata { + out[key] = value + } + return out +} + +func chunkerErrorf(format string, args ...any) error { + return fmt.Errorf("generic chunker: "+format, args...) +} diff --git a/internal/modules/chunk/generic/chunker_test.go b/internal/modules/chunk/generic/chunker_test.go new file mode 100644 index 0000000..b4a8465 --- /dev/null +++ b/internal/modules/chunk/generic/chunker_test.go @@ -0,0 +1,213 @@ +package generic + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + "strings" + "testing" + + "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 TestModuleSpecAndRegister(t *testing.T) { + want := pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageChunk, + Provides: []string{"chunks"}, + } + if got := ModuleSpec(); !reflect.DeepEqual(got, want) { + t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) + } + + registry := pipeline.NewChunkerRegistry() + if err := Register(registry); err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + spec, ok := registry.Spec(Key) + if !ok { + t.Fatalf("Spec(%q) ok = false, want true", Key) + } + if !reflect.DeepEqual(spec, want) { + t.Fatalf("registered spec = %#v, want %#v", spec, want) + } + chunker, err := registry.Build(Key) + if err != nil { + t.Fatalf("Build(%q) error = %v, want nil", Key, err) + } + if chunker.Key() != Key { + t.Fatalf("Key() = %q, want %q", chunker.Key(), Key) + } +} + +func TestChunkUsesDefaultsForSingleChunk(t *testing.T) { + result, err := New().Chunk(context.Background(), contracts.ChunkRequest{Source: testSource(3), Options: nil}) + if err != nil { + t.Fatalf("Chunk() error = %v, want nil", err) + } + + if got := chunkIDs(result.Chunks); !reflect.DeepEqual(got, []string{"chunk-000001"}) { + t.Fatalf("chunk IDs = %#v, want one stable ID", got) + } + chunk := result.Chunks[0] + if chunk.Index != 0 || chunk.SourceID != "source-1" { + t.Fatalf("chunk = %#v, want source and index fields", chunk) + } + if got := unitIDs(chunk.Units); !reflect.DeepEqual(got, []string{"u001", "u002", "u003"}) { + t.Fatalf("unit IDs = %#v, want all units", got) + } + if chunk.Metadata["start_unit_id"] != "u001" || chunk.Metadata["end_unit_id"] != "u003" || chunk.Metadata["unit_count"] != 3 { + t.Fatalf("metadata = %#v, want chunk bounds", chunk.Metadata) + } +} + +func TestChunkExactBoundaries(t *testing.T) { + result, err := New().Chunk(context.Background(), contracts.ChunkRequest{ + Source: testSource(6), + Options: map[string]any{"max_units": 2}, + }) + if err != nil { + t.Fatalf("Chunk() error = %v, want nil", err) + } + + if got := chunkIDs(result.Chunks); !reflect.DeepEqual(got, []string{"chunk-000001", "chunk-000002", "chunk-000003"}) { + t.Fatalf("chunk IDs = %#v, want stable IDs", got) + } + gotUnits := [][]string{unitIDs(result.Chunks[0].Units), unitIDs(result.Chunks[1].Units), unitIDs(result.Chunks[2].Units)} + wantUnits := [][]string{{"u001", "u002"}, {"u003", "u004"}, {"u005", "u006"}} + if !reflect.DeepEqual(gotUnits, wantUnits) { + t.Fatalf("chunk units = %#v, want %#v", gotUnits, wantUnits) + } +} + +func TestChunkOverlap(t *testing.T) { + result, err := New().Chunk(context.Background(), contracts.ChunkRequest{ + Source: testSource(7), + Options: map[string]any{"max_units": 3, "overlap_units": 1}, + }) + if err != nil { + t.Fatalf("Chunk() error = %v, want nil", err) + } + + gotUnits := make([][]string, 0, len(result.Chunks)) + for _, chunk := range result.Chunks { + gotUnits = append(gotUnits, unitIDs(chunk.Units)) + } + wantUnits := [][]string{{"u001", "u002", "u003"}, {"u003", "u004", "u005"}, {"u005", "u006", "u007"}} + if !reflect.DeepEqual(gotUnits, wantUnits) { + t.Fatalf("chunk units = %#v, want %#v", gotUnits, wantUnits) + } +} + +func TestChunkRejectsInvalidOptions(t *testing.T) { + tests := []struct { + name string + options map[string]any + want string + }{ + {name: "max wrong type", options: map[string]any{"max_units": "2"}, want: "max_units"}, + {name: "max fractional", options: map[string]any{"max_units": 1.5}, want: "integer"}, + {name: "max zero", options: map[string]any{"max_units": 0}, want: "positive"}, + {name: "overlap negative", options: map[string]any{"overlap_units": -1}, want: "non-negative"}, + {name: "overlap too large", options: map[string]any{"max_units": 2, "overlap_units": 2}, want: "less than"}, + {name: "json number", options: map[string]any{"max_units": json.Number("bad")}, want: "integer"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := New().Chunk(context.Background(), contracts.ChunkRequest{ + Source: testSource(3), + Options: test.options, + }) + if err == nil { + t.Fatal("Chunk() error = nil, want error") + } + if !strings.Contains(err.Error(), "generic chunker") || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Chunk() error = %q, want module context and %q", err.Error(), test.want) + } + }) + } +} + +func TestChunkRejectsEmptySource(t *testing.T) { + doc := testSource(1) + doc.Units = nil + + _, err := New().Chunk(context.Background(), contracts.ChunkRequest{Source: doc}) + if err == nil { + t.Fatal("Chunk() error = nil, want empty source error") + } + if !strings.Contains(err.Error(), "generic chunker") || !strings.Contains(err.Error(), "units") { + t.Fatalf("Chunk() error = %q, want empty source context", err.Error()) + } +} + +func TestChunkDefensivelyCopiesUnits(t *testing.T) { + doc := testSource(2) + + result, err := New().Chunk(context.Background(), contracts.ChunkRequest{ + Source: doc, + Options: map[string]any{"max_units": 1}, + }) + if err != nil { + t.Fatalf("Chunk() error = %v, want nil", err) + } + if len(result.Chunks) != 2 { + t.Fatalf("len(Chunks) = %d, want 2", len(result.Chunks)) + } + + doc.Units[0].ID = "changed" + doc.Units[0].Metadata["speaker"] = "changed" + + if result.Chunks[0].Units[0].ID != "u001" { + t.Fatalf("chunk unit ID changed after source mutation: %#v", result.Chunks[0].Units[0]) + } + if result.Chunks[0].Units[0].Metadata["speaker"] != "speaker-001" { + t.Fatalf("chunk unit metadata changed after source mutation: %#v", result.Chunks[0].Units[0].Metadata) + } +} + +func testSource(count int) *source.SourceDocument { + units := make([]source.SourceUnit, 0, count) + for i := 1; i <= count; i++ { + id := "u" + zeroPad3(i) + units = append(units, source.SourceUnit{ + ID: id, + Kind: "unit", + Text: "Text for " + id, + Metadata: map[string]any{ + "speaker": "speaker-" + zeroPad3(i), + }, + }) + } + return &source.SourceDocument{ + ID: "source-1", + Kind: "document", + Format: "text/plain", + Digest: "sha256:source", + Units: units, + } +} + +func zeroPad3(value int) string { + return fmt.Sprintf("%03d", value) +} + +func chunkIDs(chunks []contracts.SourceChunk) []string { + ids := make([]string, 0, len(chunks)) + for _, chunk := range chunks { + ids = append(ids, chunk.ID) + } + return ids +} + +func unitIDs(units []source.SourceUnit) []string { + ids := make([]string, 0, len(units)) + for _, unit := range units { + ids = append(ids, unit.ID) + } + return ids +} diff --git a/internal/modules/extract/dnd/spells/config_test.go b/internal/modules/extract/dnd/spells/config_test.go index b692c83..c43e276 100644 --- a/internal/modules/extract/dnd/spells/config_test.go +++ b/internal/modules/extract/dnd/spells/config_test.go @@ -11,6 +11,8 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" "gitea.maximumdirect.net/eric/notarius/internal/modules/input/seriatim" + "gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder" + "gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop" ) func TestPipelineConfigLoadsAndResolvesWithDNDSpellsExtractor(t *testing.T) { @@ -188,7 +190,7 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo Stage: pipeline.StageMerge, Requires: []string{"dnd.spell_casts"}, }, func() (contracts.Merger, error) { - return pipeline.AppendOrderMerger{}, nil + return appendorder.New(), nil }); err != nil { t.Fatalf("register merger: %v", err) } @@ -196,7 +198,7 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, }, func() (contracts.Normalizer, error) { - return pipeline.NoopNormalizer{}, nil + return noop.New(), nil }); err != nil { t.Fatalf("register normalizer: %v", err) } diff --git a/internal/modules/input/seriatim/config_test.go b/internal/modules/input/seriatim/config_test.go index acf259d..248dee8 100644 --- a/internal/modules/input/seriatim/config_test.go +++ b/internal/modules/input/seriatim/config_test.go @@ -10,6 +10,8 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/core/config" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" + "gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder" + "gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop" ) func TestPipelineConfigLoadsAndResolvesWithSeriatimInput(t *testing.T) { @@ -179,7 +181,7 @@ func mustRegisterExtractor(t *testing.T, registry *pipeline.ExtractorRegistry, s 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 + return appendorder.New(), nil }); err != nil { t.Fatalf("register merger: %v", err) } @@ -188,7 +190,7 @@ func mustRegisterMerger(t *testing.T, registry *pipeline.MergerRegistry, spec pi 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 + return noop.New(), nil }); err != nil { t.Fatalf("register normalizer: %v", err) } diff --git a/internal/modules/input/seriatim/runner_test.go b/internal/modules/input/seriatim/runner_test.go index 9a8f29b..a4be64e 100644 --- a/internal/modules/input/seriatim/runner_test.go +++ b/internal/modules/input/seriatim/runner_test.go @@ -12,6 +12,8 @@ import ( "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/merge/appendorder" + "gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop" ) func TestRunnerProcessesSeriatimInputWithFakeModules(t *testing.T) { @@ -121,12 +123,12 @@ func seriatimRunnerRegistries(t *testing.T, extractor contracts.Extractor) pipel t.Fatalf("register extractor: %v", err) } if err := mergers.Register(pipeline.DefaultMergeModule, func() (contracts.Merger, error) { - return pipeline.AppendOrderMerger{}, nil + return appendorder.New(), nil }); err != nil { t.Fatalf("register merger: %v", err) } if err := normalizers.Register(pipeline.DefaultNormalizeModule, func() (contracts.Normalizer, error) { - return pipeline.NoopNormalizer{}, nil + return noop.New(), nil }); err != nil { t.Fatalf("register normalizer: %v", err) } diff --git a/internal/modules/merge/appendorder/merger.go b/internal/modules/merge/appendorder/merger.go new file mode 100644 index 0000000..cc11351 --- /dev/null +++ b/internal/modules/merge/appendorder/merger.go @@ -0,0 +1,97 @@ +package appendorder + +import ( + "context" + "encoding/json" + "fmt" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const Key = "appendorder" + +var _ contracts.Merger = (*Merger)(nil) + +type Merger struct{} + +func New() *Merger { + return &Merger{} +} + +func (m *Merger) Key() string { + return Key +} + +func (m *Merger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) { + if m == nil { + return contracts.MergeResult{}, mergerErrorf("merger must not be nil") + } + if ctx == nil { + return contracts.MergeResult{}, mergerErrorf("context must not be nil") + } + if err := ctx.Err(); err != nil { + return contracts.MergeResult{}, mergerErrorf("context error before merge: %w", err) + } + + var candidates []artifacts.ArtifactCandidate + for _, chunkArtifacts := range req.ChunkArtifacts { + candidates = append(candidates, cloneCandidates(chunkArtifacts.Candidates)...) + } + return contracts.MergeResult{Candidates: candidates}, nil +} + +func ModuleSpec() pipeline.ModuleSpec { + return pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageMerge, + Provides: []string{"merged"}, + } +} + +func Register(registry *pipeline.MergerRegistry) error { + return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.Merger, error) { + return New(), nil + }) +} + +func cloneCandidates(candidates []artifacts.ArtifactCandidate) []artifacts.ArtifactCandidate { + if len(candidates) == 0 { + return nil + } + + out := make([]artifacts.ArtifactCandidate, 0, len(candidates)) + for _, candidate := range candidates { + out = append(out, cloneCandidate(candidate)) + } + return out +} + +func cloneCandidate(candidate artifacts.ArtifactCandidate) artifacts.ArtifactCandidate { + return artifacts.ArtifactCandidate{ + Index: candidate.Index, + ExtractorKey: candidate.ExtractorKey, + ArtifactType: candidate.ArtifactType, + SchemaVersion: candidate.SchemaVersion, + Payload: append(json.RawMessage(nil), candidate.Payload...), + SourceRefs: append([]source.SourceRef(nil), candidate.SourceRefs...), + Metadata: cloneMetadata(candidate.Metadata), + } +} + +func cloneMetadata(metadata map[string]any) map[string]any { + if len(metadata) == 0 { + return nil + } + out := make(map[string]any, len(metadata)) + for key, value := range metadata { + out[key] = value + } + return out +} + +func mergerErrorf(format string, args ...any) error { + return fmt.Errorf("appendorder merger: "+format, args...) +} diff --git a/internal/modules/merge/appendorder/merger_test.go b/internal/modules/merge/appendorder/merger_test.go new file mode 100644 index 0000000..0b04a80 --- /dev/null +++ b/internal/modules/merge/appendorder/merger_test.go @@ -0,0 +1,147 @@ +package appendorder + +import ( + "context" + "encoding/json" + "reflect" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "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 TestModuleSpecAndRegister(t *testing.T) { + want := pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageMerge, + Provides: []string{"merged"}, + } + if got := ModuleSpec(); !reflect.DeepEqual(got, want) { + t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) + } + + registry := pipeline.NewMergerRegistry() + if err := Register(registry); err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + spec, ok := registry.Spec(Key) + if !ok { + t.Fatalf("Spec(%q) ok = false, want true", Key) + } + if !reflect.DeepEqual(spec, want) { + t.Fatalf("registered spec = %#v, want %#v", spec, want) + } +} + +func TestMergePreservesChunkAndCandidateOrder(t *testing.T) { + result, err := New().Merge(context.Background(), contracts.MergeRequest{ + ChunkArtifacts: []contracts.ChunkArtifacts{ + { + Chunk: sourceChunk(0), + Candidates: []artifacts.ArtifactCandidate{candidate(2, "first-b"), candidate(1, "first-a")}, + }, + { + Chunk: sourceChunk(1), + Candidates: []artifacts.ArtifactCandidate{candidate(4, "second-b"), candidate(3, "second-a")}, + }, + }, + }) + if err != nil { + t.Fatalf("Merge() error = %v, want nil", err) + } + + got := candidateNames(result.Candidates) + want := []string{"first-b", "first-a", "second-b", "second-a"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("candidate order = %#v, want %#v", got, want) + } + if len(result.Warnings) != 0 { + t.Fatalf("Warnings = %#v, want none", result.Warnings) + } +} + +func TestMergeDefensivelyCopiesCandidates(t *testing.T) { + input := []contracts.ChunkArtifacts{ + { + Chunk: sourceChunk(0), + Candidates: []artifacts.ArtifactCandidate{candidate(1, "original")}, + }, + } + + result, err := New().Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: input}) + if err != nil { + t.Fatalf("Merge() error = %v, want nil", err) + } + if len(result.Candidates) != 1 { + t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates)) + } + + input[0].Candidates[0].Index = 99 + input[0].Candidates[0].Payload[0] = '[' + input[0].Candidates[0].SourceRefs[0].StartUnitID = "changed" + input[0].Candidates[0].Metadata["name"] = "changed" + + got := result.Candidates[0] + if got.Index != 1 { + t.Fatalf("Index = %d, want 1", got.Index) + } + if string(got.Payload) != `{"name":"original"}` { + t.Fatalf("Payload = %s, want original payload", got.Payload) + } + if got.SourceRefs[0].StartUnitID != "u1" { + t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs) + } + if got.Metadata["name"] != "original" { + t.Fatalf("Metadata = %#v, want original metadata", got.Metadata) + } +} + +func TestMergeHandlesEmptyInput(t *testing.T) { + result, err := New().Merge(context.Background(), contracts.MergeRequest{}) + if err != nil { + t.Fatalf("Merge() error = %v, want nil", err) + } + if len(result.Candidates) != 0 { + t.Fatalf("len(Candidates) = %d, want 0", len(result.Candidates)) + } + if len(result.Warnings) != 0 { + t.Fatalf("Warnings = %#v, want none", result.Warnings) + } +} + +func candidate(index int, name string) artifacts.ArtifactCandidate { + return artifacts.ArtifactCandidate{ + Index: index, + ExtractorKey: "generic-extractor", + ArtifactType: "generic-artifact", + SchemaVersion: "v1", + Payload: json.RawMessage(`{"name":"` + name + `"}`), + SourceRefs: []source.SourceRef{ + {SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"}, + }, + Metadata: map[string]any{ + "name": name, + }, + } +} + +func candidateNames(candidates []artifacts.ArtifactCandidate) []string { + names := make([]string, 0, len(candidates)) + for _, candidate := range candidates { + names = append(names, candidate.Metadata["name"].(string)) + } + return names +} + +func sourceChunk(index int) contracts.SourceChunk { + return contracts.SourceChunk{ + ID: "chunk", + SourceID: "source-1", + Index: index, + Units: []source.SourceUnit{ + {ID: "u1", Kind: "unit", Text: "Source unit."}, + }, + } +} diff --git a/internal/modules/normalize/noop/normalizer.go b/internal/modules/normalize/noop/normalizer.go new file mode 100644 index 0000000..4016e58 --- /dev/null +++ b/internal/modules/normalize/noop/normalizer.go @@ -0,0 +1,89 @@ +package noop + +import ( + "context" + "encoding/json" + "fmt" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const Key = "noop" + +var _ contracts.Normalizer = (*Normalizer)(nil) + +type Normalizer struct{} + +func New() *Normalizer { + return &Normalizer{} +} + +func (n *Normalizer) Key() string { + return Key +} + +func (n *Normalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) { + if n == nil { + return contracts.NormalizeResult{}, normalizerErrorf("normalizer must not be nil") + } + if ctx == nil { + return contracts.NormalizeResult{}, normalizerErrorf("context must not be nil") + } + if err := ctx.Err(); err != nil { + return contracts.NormalizeResult{}, normalizerErrorf("context error before normalize: %w", err) + } + return contracts.NormalizeResult{Candidates: cloneCandidates(req.Candidates)}, nil +} + +func ModuleSpec() pipeline.ModuleSpec { + return pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + } +} + +func Register(registry *pipeline.NormalizerRegistry) error { + return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.Normalizer, error) { + return New(), nil + }) +} + +func cloneCandidates(candidates []artifacts.ArtifactCandidate) []artifacts.ArtifactCandidate { + if len(candidates) == 0 { + return nil + } + + out := make([]artifacts.ArtifactCandidate, 0, len(candidates)) + for _, candidate := range candidates { + out = append(out, artifacts.ArtifactCandidate{ + Index: candidate.Index, + ExtractorKey: candidate.ExtractorKey, + ArtifactType: candidate.ArtifactType, + SchemaVersion: candidate.SchemaVersion, + Payload: append(json.RawMessage(nil), candidate.Payload...), + SourceRefs: append([]source.SourceRef(nil), candidate.SourceRefs...), + Metadata: cloneMetadata(candidate.Metadata), + }) + } + return out +} + +func cloneMetadata(metadata map[string]any) map[string]any { + if len(metadata) == 0 { + return nil + } + out := make(map[string]any, len(metadata)) + for key, value := range metadata { + out[key] = value + } + return out +} + +func normalizerErrorf(format string, args ...any) error { + return fmt.Errorf("noop normalizer: "+format, args...) +} diff --git a/internal/modules/normalize/noop/normalizer_test.go b/internal/modules/normalize/noop/normalizer_test.go new file mode 100644 index 0000000..61445da --- /dev/null +++ b/internal/modules/normalize/noop/normalizer_test.go @@ -0,0 +1,133 @@ +package noop + +import ( + "context" + "encoding/json" + "reflect" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "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 TestModuleSpecAndRegister(t *testing.T) { + want := pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + } + if got := ModuleSpec(); !reflect.DeepEqual(got, want) { + t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) + } + + registry := pipeline.NewNormalizerRegistry() + if err := Register(registry); err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + spec, ok := registry.Spec(Key) + if !ok { + t.Fatalf("Spec(%q) ok = false, want true", Key) + } + if !reflect.DeepEqual(spec, want) { + t.Fatalf("registered spec = %#v, want %#v", spec, want) + } +} + +func TestNormalizePassesThroughOrderAndValues(t *testing.T) { + input := []artifacts.ArtifactCandidate{ + candidate(3, "third"), + candidate(1, "first"), + candidate(2, "second"), + } + + result, err := New().Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input}) + if err != nil { + t.Fatalf("Normalize() error = %v, want nil", err) + } + if len(result.Warnings) != 0 { + t.Fatalf("Warnings = %#v, want none", result.Warnings) + } + + got := candidateNames(result.Candidates) + want := []string{"third", "first", "second"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("candidate order = %#v, want %#v", got, want) + } + if !reflect.DeepEqual(result.Candidates[0].SourceRefs, input[0].SourceRefs) { + t.Fatalf("SourceRefs = %#v, want %#v", result.Candidates[0].SourceRefs, input[0].SourceRefs) + } + if !reflect.DeepEqual(result.Candidates[0].Metadata, input[0].Metadata) { + t.Fatalf("Metadata = %#v, want %#v", result.Candidates[0].Metadata, input[0].Metadata) + } +} + +func TestNormalizeDefensivelyCopiesCandidates(t *testing.T) { + input := []artifacts.ArtifactCandidate{candidate(1, "original")} + + result, err := New().Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input}) + if err != nil { + t.Fatalf("Normalize() error = %v, want nil", err) + } + if len(result.Candidates) != 1 { + t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates)) + } + + input[0].Index = 99 + input[0].Payload[0] = '[' + input[0].SourceRefs[0].EndUnitID = "changed" + input[0].Metadata["name"] = "changed" + + got := result.Candidates[0] + if got.Index != 1 { + t.Fatalf("Index = %d, want 1", got.Index) + } + if string(got.Payload) != `{"name":"original"}` { + t.Fatalf("Payload = %s, want original payload", got.Payload) + } + if got.SourceRefs[0].EndUnitID != "u1" { + t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs) + } + if got.Metadata["name"] != "original" { + t.Fatalf("Metadata = %#v, want original metadata", got.Metadata) + } +} + +func TestNormalizeHandlesEmptyInput(t *testing.T) { + result, err := New().Normalize(context.Background(), contracts.NormalizeRequest{}) + if err != nil { + t.Fatalf("Normalize() error = %v, want nil", err) + } + if len(result.Candidates) != 0 { + t.Fatalf("len(Candidates) = %d, want 0", len(result.Candidates)) + } + if len(result.Warnings) != 0 { + t.Fatalf("Warnings = %#v, want none", result.Warnings) + } +} + +func candidate(index int, name string) artifacts.ArtifactCandidate { + return artifacts.ArtifactCandidate{ + Index: index, + ExtractorKey: "generic-extractor", + ArtifactType: "generic-artifact", + SchemaVersion: "v1", + Payload: json.RawMessage(`{"name":"` + name + `"}`), + SourceRefs: []source.SourceRef{ + {SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"}, + }, + Metadata: map[string]any{ + "name": name, + }, + } +} + +func candidateNames(candidates []artifacts.ArtifactCandidate) []string { + names := make([]string, 0, len(candidates)) + for _, candidate := range candidates { + names = append(names, candidate.Metadata["name"].(string)) + } + return names +} diff --git a/internal/modules/output/json/encoder.go b/internal/modules/output/json/encoder.go new file mode 100644 index 0000000..1eb80e6 --- /dev/null +++ b/internal/modules/output/json/encoder.go @@ -0,0 +1,253 @@ +package json + +import ( + "context" + stdjson "encoding/json" + "fmt" + "regexp" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const Key = "json" + +const contentTypeJSON = "application/json" + +var safeArtifactFileChar = regexp.MustCompile(`[^A-Za-z0-9._-]`) + +var _ contracts.OutputEncoder = (*Encoder)(nil) + +type Encoder struct{} + +func New() *Encoder { + return &Encoder{} +} + +func (e *Encoder) Key() string { + return Key +} + +func (e *Encoder) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) { + if e == nil { + return contracts.OutputResult{}, encoderErrorf("encoder must not be nil") + } + if ctx == nil { + return contracts.OutputResult{}, encoderErrorf("context must not be nil") + } + if err := ctx.Err(); err != nil { + return contracts.OutputResult{}, encoderErrorf("context error before encoding: %w", err) + } + + files, err := logicalFiles(req) + if err != nil { + return contracts.OutputResult{}, err + } + return contracts.OutputResult{Files: files}, nil +} + +func ModuleSpec() pipeline.ModuleSpec { + return pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageOutput, + Requires: []string{"normalized"}, + Provides: []string{"encoded"}, + } +} + +func Register(registry *pipeline.OutputEncoderRegistry) error { + return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.OutputEncoder, error) { + return New(), nil + }) +} + +type indexFile struct { + ManifestFile string `json:"manifest_file"` + ArtifactFiles []artifactFileIndex `json:"artifact_files"` + RejectedFile string `json:"rejected_file"` + WarningsFile string `json:"warnings_file"` +} + +type artifactFileIndex struct { + ArtifactType string `json:"artifact_type"` + File string `json:"file"` +} + +type artifactFile struct { + ArtifactType string `json:"artifact_type"` + Artifacts []artifacts.Artifact `json:"artifacts"` +} + +type rejectedFile struct { + Rejected []artifacts.RejectedArtifact `json:"rejected"` +} + +type warningsFile struct { + Warnings []contracts.Warning `json:"warnings"` +} + +func logicalFiles(req contracts.OutputRequest) ([]contracts.OutputFile, error) { + artifactsByType := make(map[string][]artifacts.Artifact) + for _, artifact := range req.Approved { + artifactsByType[artifact.ArtifactType] = append(artifactsByType[artifact.ArtifactType], cloneArtifact(artifact)) + } + + artifactTypes := make([]string, 0, len(artifactsByType)) + for artifactType := range artifactsByType { + artifactTypes = append(artifactTypes, artifactType) + } + sort.Strings(artifactTypes) + + artifactIndexes := make([]artifactFileIndex, 0, len(artifactTypes)) + files := make([]contracts.OutputFile, 0, len(artifactTypes)+4) + manifestFile, err := jsonFile("manifest.json", req.Manifest) + if err != nil { + return nil, err + } + files = append(files, manifestFile) + + usedArtifactFiles := make(map[string]string, len(artifactTypes)) + for _, artifactType := range artifactTypes { + name, err := artifactFileName(artifactType) + if err != nil { + return nil, err + } + if existingType, ok := usedArtifactFiles[name]; ok { + return nil, encoderErrorf("artifact types %q and %q produce duplicate output file %q", existingType, artifactType, name) + } + usedArtifactFiles[name] = artifactType + artifactIndexes = append(artifactIndexes, artifactFileIndex{ + ArtifactType: artifactType, + File: name, + }) + file, err := jsonFile(name, artifactFile{ + ArtifactType: artifactType, + Artifacts: artifactsByType[artifactType], + }) + if err != nil { + return nil, err + } + files = append(files, file) + } + + index := indexFile{ + ManifestFile: "manifest.json", + ArtifactFiles: artifactIndexes, + RejectedFile: "rejected.json", + WarningsFile: "warnings.json", + } + indexOutput, err := jsonFile("index.json", index) + if err != nil { + return nil, err + } + rejectedOutput, err := jsonFile("rejected.json", rejectedFile{Rejected: cloneRejected(req.Rejected)}) + if err != nil { + return nil, err + } + warningsOutput, err := jsonFile("warnings.json", warningsFile{Warnings: cloneWarnings(req.Warnings)}) + if err != nil { + return nil, err + } + files = append(files, indexOutput, rejectedOutput, warningsOutput) + sort.Slice(files, func(i, j int) bool { + return files[i].Name < files[j].Name + }) + return files, nil +} + +func jsonFile(name string, value any) (contracts.OutputFile, error) { + data, err := marshalPretty(value) + if err != nil { + return contracts.OutputFile{}, encoderErrorf("encode %s: %w", name, err) + } + return contracts.OutputFile{ + Name: name, + ContentType: contentTypeJSON, + Bytes: data, + }, nil +} + +func marshalPretty(value any) ([]byte, error) { + data, err := stdjson.MarshalIndent(value, "", " ") + if err != nil { + return nil, err + } + return append(data, '\n'), nil +} + +func artifactFileName(artifactType string) (string, error) { + sanitized := safeArtifactFileChar.ReplaceAllString(strings.TrimSpace(artifactType), "_") + for strings.Contains(sanitized, "..") { + sanitized = strings.ReplaceAll(sanitized, "..", "__") + } + sanitized = strings.Trim(sanitized, "._") + if sanitized == "" { + return "", encoderErrorf("artifact type %q cannot produce a safe file name", artifactType) + } + return "artifacts/" + sanitized + ".json", nil +} + +func cloneArtifact(artifact artifacts.Artifact) artifacts.Artifact { + return artifacts.Artifact{ + ExtractorKey: artifact.ExtractorKey, + ArtifactType: artifact.ArtifactType, + SchemaVersion: artifact.SchemaVersion, + Payload: append(stdjson.RawMessage(nil), artifact.Payload...), + SourceRefs: append([]source.SourceRef(nil), artifact.SourceRefs...), + Metadata: cloneMetadata(artifact.Metadata), + } +} + +func cloneRejected(rejected []artifacts.RejectedArtifact) []artifacts.RejectedArtifact { + if len(rejected) == 0 { + return []artifacts.RejectedArtifact{} + } + out := make([]artifacts.RejectedArtifact, 0, len(rejected)) + for _, item := range rejected { + out = append(out, artifacts.RejectedArtifact{ + Candidate: cloneCandidate(item.Candidate), + ValidatorName: item.ValidatorName, + ReasonCode: item.ReasonCode, + Message: item.Message, + }) + } + return out +} + +func cloneCandidate(candidate artifacts.ArtifactCandidate) artifacts.ArtifactCandidate { + return artifacts.ArtifactCandidate{ + Index: candidate.Index, + ExtractorKey: candidate.ExtractorKey, + ArtifactType: candidate.ArtifactType, + SchemaVersion: candidate.SchemaVersion, + Payload: append(stdjson.RawMessage(nil), candidate.Payload...), + SourceRefs: append([]source.SourceRef(nil), candidate.SourceRefs...), + Metadata: cloneMetadata(candidate.Metadata), + } +} + +func cloneWarnings(warnings []contracts.Warning) []contracts.Warning { + if len(warnings) == 0 { + return []contracts.Warning{} + } + return append([]contracts.Warning(nil), warnings...) +} + +func cloneMetadata(metadata map[string]any) map[string]any { + if len(metadata) == 0 { + return nil + } + out := make(map[string]any, len(metadata)) + for key, value := range metadata { + out[key] = value + } + return out +} + +func encoderErrorf(format string, args ...any) error { + return fmt.Errorf("json output encoder: "+format, args...) +} diff --git a/internal/modules/output/json/encoder_test.go b/internal/modules/output/json/encoder_test.go new file mode 100644 index 0000000..c0c9c83 --- /dev/null +++ b/internal/modules/output/json/encoder_test.go @@ -0,0 +1,316 @@ +package json + +import ( + "context" + stdjson "encoding/json" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "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 TestModuleSpecAndRegister(t *testing.T) { + want := pipeline.ModuleSpec{ + Key: Key, + Stage: pipeline.StageOutput, + Requires: []string{"normalized"}, + Provides: []string{"encoded"}, + } + if got := ModuleSpec(); !reflect.DeepEqual(got, want) { + t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) + } + + registry := pipeline.NewOutputEncoderRegistry() + if err := Register(registry); err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + spec, ok := registry.Spec(Key) + if !ok { + t.Fatalf("Spec(%q) ok = false, want true", Key) + } + if !reflect.DeepEqual(spec, want) { + t.Fatalf("registered spec = %#v, want %#v", spec, want) + } +} + +func TestEncodeReturnsLogicalFilesGroupedByArtifactType(t *testing.T) { + req := contracts.OutputRequest{ + Manifest: artifacts.RunManifest{RunID: "run-1", PipelineID: "pipeline-1"}, + Approved: []artifacts.Artifact{ + artifact("dnd.spell-cast", "first"), + artifact("notes/item", "item"), + artifact("dnd.spell-cast", "second"), + }, + Rejected: []artifacts.RejectedArtifact{ + { + Candidate: candidate("bad type", "bad"), + ValidatorName: "validator", + ReasonCode: "invalid", + Message: "not accepted", + }, + }, + Warnings: []contracts.Warning{{ReasonCode: "warning", Message: "check source"}}, + } + + result, err := New().Encode(context.Background(), req) + if err != nil { + t.Fatalf("Encode() error = %v, want nil", err) + } + + wantNames := []string{ + "artifacts/dnd.spell-cast.json", + "artifacts/notes_item.json", + "index.json", + "manifest.json", + "rejected.json", + "warnings.json", + } + if got := outputFileNames(result.Files); !reflect.DeepEqual(got, wantNames) { + t.Fatalf("file names = %#v, want %#v", got, wantNames) + } + for _, file := range result.Files { + if file.ContentType != contentTypeJSON { + t.Fatalf("%s ContentType = %q, want %q", file.Name, file.ContentType, contentTypeJSON) + } + if !strings.HasSuffix(string(file.Bytes), "\n") { + t.Fatalf("%s does not end with newline: %q", file.Name, string(file.Bytes)) + } + if !stdjson.Valid(file.Bytes) { + t.Fatalf("%s has invalid JSON: %s", file.Name, file.Bytes) + } + } + + spellFile := decodeObject(t, fileBytes(t, result.Files, "artifacts/dnd.spell-cast.json")) + if spellFile["artifact_type"] != "dnd.spell-cast" { + t.Fatalf("artifact_type = %#v, want dnd.spell-cast", spellFile["artifact_type"]) + } + spells := spellFile["artifacts"].([]any) + if len(spells) != 2 { + t.Fatalf("len(spells) = %d, want 2", len(spells)) + } + firstPayload := spells[0].(map[string]any)["payload"].(map[string]any) + secondPayload := spells[1].(map[string]any)["payload"].(map[string]any) + if firstPayload["name"] != "first" || secondPayload["name"] != "second" { + t.Fatalf("spell order payloads = %#v then %#v, want runner order", firstPayload, secondPayload) + } + + index := decodeObject(t, fileBytes(t, result.Files, "index.json")) + artifactFiles := index["artifact_files"].([]any) + if len(artifactFiles) != 2 { + t.Fatalf("len(index artifact_files) = %d, want 2", len(artifactFiles)) + } + firstIndex := artifactFiles[0].(map[string]any) + secondIndex := artifactFiles[1].(map[string]any) + if firstIndex["artifact_type"] != "dnd.spell-cast" || secondIndex["artifact_type"] != "notes/item" { + t.Fatalf("artifact_files = %#v, want sorted by artifact type", artifactFiles) + } +} + +func TestEncodeIncludesRejectedAndWarningsWhenEmpty(t *testing.T) { + result, err := New().Encode(context.Background(), contracts.OutputRequest{ + Manifest: artifacts.RunManifest{RunID: "run-1"}, + }) + if err != nil { + t.Fatalf("Encode() error = %v, want nil", err) + } + + rejected := decodeObject(t, fileBytes(t, result.Files, "rejected.json")) + if got := rejected["rejected"].([]any); len(got) != 0 { + t.Fatalf("rejected = %#v, want empty array", got) + } + warnings := decodeObject(t, fileBytes(t, result.Files, "warnings.json")) + if got := warnings["warnings"].([]any); len(got) != 0 { + t.Fatalf("warnings = %#v, want empty array", got) + } +} + +func TestEncodePrettyPrintsJSON(t *testing.T) { + result, err := New().Encode(context.Background(), contracts.OutputRequest{ + Manifest: artifacts.RunManifest{RunID: "run-1"}, + }) + if err != nil { + t.Fatalf("Encode() error = %v, want nil", err) + } + + manifest := string(fileBytes(t, result.Files, "manifest.json")) + if !strings.Contains(manifest, "\n \"run_id\": \"run-1\"\n") { + t.Fatalf("manifest JSON = %q, want two-space indentation", manifest) + } +} + +func TestEncodeRejectsArtifactTypeWithoutSafeFileName(t *testing.T) { + _, err := New().Encode(context.Background(), contracts.OutputRequest{ + Approved: []artifacts.Artifact{artifact("///", "unsafe")}, + }) + if err == nil { + t.Fatal("Encode() error = nil, want unsafe artifact type error") + } + if !strings.Contains(err.Error(), "json output encoder") || !strings.Contains(err.Error(), "safe file name") { + t.Fatalf("Encode() error = %q, want safe file name context", err.Error()) + } +} + +func TestEncodeSanitizesParentPathSequences(t *testing.T) { + result, err := New().Encode(context.Background(), contracts.OutputRequest{ + Approved: []artifacts.Artifact{artifact("dnd..spell.", "spell")}, + }) + if err != nil { + t.Fatalf("Encode() error = %v, want nil", err) + } + + if got := outputFileNames(result.Files); !containsString(got, "artifacts/dnd__spell.json") { + t.Fatalf("file names = %#v, want sanitized artifact filename", got) + } +} + +func TestEncodeRejectsSanitizedFilenameCollisions(t *testing.T) { + _, err := New().Encode(context.Background(), contracts.OutputRequest{ + Approved: []artifacts.Artifact{ + artifact("a/b", "slash"), + artifact("a?b", "question"), + }, + }) + if err == nil { + t.Fatal("Encode() error = nil, want duplicate file error") + } + if !strings.Contains(err.Error(), "duplicate output file") { + t.Fatalf("Encode() error = %q, want duplicate file context", err.Error()) + } +} + +func TestEncodeDoesNotMutateInputs(t *testing.T) { + req := contracts.OutputRequest{ + Manifest: artifacts.RunManifest{RunID: "run-1"}, + Approved: []artifacts.Artifact{ + artifact("dnd.spell", "original"), + }, + Rejected: []artifacts.RejectedArtifact{ + { + Candidate: candidate("bad", "rejected"), + ValidatorName: "validator", + ReasonCode: "invalid", + Message: "not accepted", + }, + }, + Warnings: []contracts.Warning{{ReasonCode: "warning", Message: "message"}}, + } + before := mustMarshal(t, req) + + result, err := New().Encode(context.Background(), req) + if err != nil { + t.Fatalf("Encode() error = %v, want nil", err) + } + after := mustMarshal(t, req) + if before != after { + t.Fatalf("request mutated:\nbefore: %s\nafter: %s", before, after) + } + + req.Approved[0].Payload[0] = '[' + req.Approved[0].SourceRefs[0].StartUnitID = "changed" + req.Approved[0].Metadata["name"] = "changed" + req.Rejected[0].Candidate.Payload[0] = '[' + req.Warnings[0].Message = "changed" + + if !stdjson.Valid(fileBytes(t, result.Files, "artifacts/dnd.spell.json")) { + t.Fatal("artifact output changed after request mutation") + } + warnings := decodeObject(t, fileBytes(t, result.Files, "warnings.json")) + gotWarnings := warnings["warnings"].([]any) + if gotWarnings[0].(map[string]any)["message"] != "message" { + t.Fatalf("warnings output changed after request mutation: %#v", gotWarnings) + } +} + +func TestArtifactFilesDoNotContainWarnings(t *testing.T) { + result, err := New().Encode(context.Background(), contracts.OutputRequest{ + Approved: []artifacts.Artifact{artifact("dnd.spell", "spell")}, + Warnings: []contracts.Warning{ + {ReasonCode: "pipeline-warning", Message: "warning"}, + }, + }) + if err != nil { + t.Fatalf("Encode() error = %v, want nil", err) + } + + artifactFile := decodeObject(t, fileBytes(t, result.Files, "artifacts/dnd.spell.json")) + if _, ok := artifactFile["warnings"]; ok { + t.Fatalf("artifact file contains warnings: %#v", artifactFile) + } +} + +func artifact(artifactType, name string) artifacts.Artifact { + return artifacts.Artifact{ + ExtractorKey: "extractor", + ArtifactType: artifactType, + SchemaVersion: "v1", + Payload: stdjson.RawMessage(`{"name":"` + name + `"}`), + SourceRefs: []source.SourceRef{ + {SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"}, + }, + Metadata: map[string]any{"name": name}, + } +} + +func candidate(artifactType, name string) artifacts.ArtifactCandidate { + return artifacts.ArtifactCandidate{ + Index: 1, + ExtractorKey: "extractor", + ArtifactType: artifactType, + SchemaVersion: "v1", + Payload: stdjson.RawMessage(`{"name":"` + name + `"}`), + SourceRefs: []source.SourceRef{ + {SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"}, + }, + Metadata: map[string]any{"name": name}, + } +} + +func outputFileNames(files []contracts.OutputFile) []string { + names := make([]string, 0, len(files)) + for _, file := range files { + names = append(names, file.Name) + } + return names +} + +func containsString(values []string, want string) bool { + for _, value := range values { + if value == want { + return true + } + } + return false +} + +func fileBytes(t *testing.T, files []contracts.OutputFile, name string) []byte { + t.Helper() + for _, file := range files { + if file.Name == name { + return file.Bytes + } + } + t.Fatalf("file %q not found in %#v", name, outputFileNames(files)) + return nil +} + +func decodeObject(t *testing.T, data []byte) map[string]any { + t.Helper() + var got map[string]any + if err := stdjson.Unmarshal(data, &got); err != nil { + t.Fatalf("Unmarshal() error = %v, want nil\n%s", err, data) + } + return got +} + +func mustMarshal(t *testing.T, value any) string { + t.Helper() + data, err := stdjson.Marshal(value) + if err != nil { + t.Fatalf("Marshal() error = %v, want nil", err) + } + return string(data) +}