From 8217561f4f889a1a17e04f1e0cad0b6a029d2fab Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 3 Jul 2026 15:39:10 +0000 Subject: [PATCH] Add generic merge and normalize helpers --- internal/framework/pipeline/generic_stages.go | 70 ++++++ .../framework/pipeline/generic_stages_test.go | 221 ++++++++++++++++++ 2 files changed, 291 insertions(+) create mode 100644 internal/framework/pipeline/generic_stages.go create mode 100644 internal/framework/pipeline/generic_stages_test.go diff --git a/internal/framework/pipeline/generic_stages.go b/internal/framework/pipeline/generic_stages.go new file mode 100644 index 0000000..340d8db --- /dev/null +++ b/internal/framework/pipeline/generic_stages.go @@ -0,0 +1,70 @@ +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 new file mode 100644 index 0000000..400ffde --- /dev/null +++ b/internal/framework/pipeline/generic_stages_test.go @@ -0,0 +1,221 @@ +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."}, + }, + } +}