290 lines
8.5 KiB
Go
290 lines
8.5 KiB
Go
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"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"
|
|
|
|
validate "gitea.maximumdirect.net/eric/notarius/internal/framework/validate"
|
|
)
|
|
|
|
func TestRunnerUsesRegistries(t *testing.T) {
|
|
var built []string
|
|
var executed []string
|
|
registries := integrationRegistries(t, &built, &executed)
|
|
|
|
output, err := New(registries).Run(context.Background(), RunInput{
|
|
Pipeline: integrationPipeline(),
|
|
SourceID: "source-1",
|
|
RawInput: []byte("source text"),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("Run() error = %v, want nil", err)
|
|
}
|
|
|
|
wantBuilt := []string{"input", "chunk", "extract-first", "merge", "normalize", "extract-second", "merge", "normalize", "output"}
|
|
if !reflect.DeepEqual(built, wantBuilt) {
|
|
t.Fatalf("built = %#v, want %#v", built, wantBuilt)
|
|
}
|
|
if !reflect.DeepEqual(executed, []string{"extract-first:chunk-0", "extract-second:chunk-0"}) {
|
|
t.Fatalf("executed = %#v, want extractor chunk execution", executed)
|
|
}
|
|
if got := artifactKeys(output.Approved); !reflect.DeepEqual(got, []string{"extract-first"}) {
|
|
t.Fatalf("approved keys = %#v, want [extract-first]", got)
|
|
}
|
|
if got := rejectedKeys(output.Rejected); !reflect.DeepEqual(got, []string{"extract-second"}) {
|
|
t.Fatalf("rejected keys = %#v, want [extract-second]", got)
|
|
}
|
|
}
|
|
|
|
func integrationRegistries(t *testing.T, built, executed *[]string) Registries {
|
|
t.Helper()
|
|
|
|
registries := Registries{
|
|
Inputs: NewInputAdapterRegistry(),
|
|
Chunkers: NewChunkerRegistry(),
|
|
Extractors: NewExtractorRegistry(),
|
|
Mergers: NewMergerRegistry(),
|
|
Normalizers: NewNormalizerRegistry(),
|
|
Outputs: NewOutputEncoderRegistry(),
|
|
}
|
|
if err := registries.Inputs.Register("input", func() (contracts.InputAdapter, error) {
|
|
*built = append(*built, "input")
|
|
return integrationInput{}, nil
|
|
}); err != nil {
|
|
t.Fatalf("register input: %v", err)
|
|
}
|
|
if err := registries.Chunkers.Register("chunk", func() (contracts.Chunker, error) {
|
|
*built = append(*built, "chunk")
|
|
return integrationChunker{}, nil
|
|
}); err != nil {
|
|
t.Fatalf("register chunker: %v", err)
|
|
}
|
|
registerIntegrationExtractor(t, registries.Extractors, "extract-first", built, executed, []contracts.Validator{
|
|
integrationValidator{name: "approve-first", approve: true},
|
|
})
|
|
registerIntegrationExtractor(t, registries.Extractors, "extract-second", built, executed, []contracts.Validator{
|
|
integrationValidator{name: "reject-second", approve: false},
|
|
})
|
|
if err := registries.Mergers.Register("merge", func() (contracts.Merger, error) {
|
|
*built = append(*built, "merge")
|
|
return integrationMerger{}, nil
|
|
}); err != nil {
|
|
t.Fatalf("register merger: %v", err)
|
|
}
|
|
if err := registries.Normalizers.Register("normalize", func() (contracts.Normalizer, error) {
|
|
*built = append(*built, "normalize")
|
|
return integrationNormalizer{}, nil
|
|
}); err != nil {
|
|
t.Fatalf("register normalizer: %v", err)
|
|
}
|
|
if err := registries.Outputs.Register("output", func() (contracts.OutputEncoder, error) {
|
|
*built = append(*built, "output")
|
|
return integrationOutput{}, nil
|
|
}); err != nil {
|
|
t.Fatalf("register output: %v", err)
|
|
}
|
|
return registries
|
|
}
|
|
|
|
func registerIntegrationExtractor(t *testing.T, registry *ExtractorRegistry, key string, built, executed *[]string, validators []contracts.Validator) {
|
|
t.Helper()
|
|
|
|
if err := registry.Register(key, func() (contracts.Extractor, error) {
|
|
*built = append(*built, key)
|
|
return integrationExtractor{key: key, executed: executed, validators: validators}, nil
|
|
}); err != nil {
|
|
t.Fatalf("Register(%q) error = %v, want nil", key, err)
|
|
}
|
|
}
|
|
|
|
type integrationInput struct{}
|
|
|
|
func (input integrationInput) Key() string {
|
|
return "input"
|
|
}
|
|
|
|
func (input integrationInput) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
|
|
return integrationSourceDocument(), nil
|
|
}
|
|
|
|
type integrationChunker struct{}
|
|
|
|
func (chunker integrationChunker) Key() string {
|
|
return "chunk"
|
|
}
|
|
|
|
func (chunker integrationChunker) ReferenceSlots() []contracts.ReferenceSlot {
|
|
return nil
|
|
}
|
|
|
|
func (chunker integrationChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
|
|
return contracts.ChunkResult{
|
|
Chunks: []contracts.SourceChunk{
|
|
{
|
|
ID: "chunk-0",
|
|
SourceID: req.Source.ID,
|
|
Index: 0,
|
|
Units: req.Source.Units,
|
|
},
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
type integrationExtractor struct {
|
|
key string
|
|
executed *[]string
|
|
validators []contracts.Validator
|
|
}
|
|
|
|
func (extractor integrationExtractor) Key() string {
|
|
return extractor.key
|
|
}
|
|
|
|
func (extractor integrationExtractor) ArtifactType() string {
|
|
return "generic-artifact"
|
|
}
|
|
|
|
func (extractor integrationExtractor) SchemaVersion() string {
|
|
return "v1"
|
|
}
|
|
|
|
func (extractor integrationExtractor) ReferenceSlots() []contracts.ReferenceSlot {
|
|
return nil
|
|
}
|
|
|
|
func (extractor integrationExtractor) Validators() []contracts.Validator {
|
|
return extractor.validators
|
|
}
|
|
|
|
func (extractor integrationExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) {
|
|
*extractor.executed = append(*extractor.executed, extractor.key+":"+req.Chunk.ID)
|
|
return contracts.ExtractionResult{
|
|
Candidates: []artifacts.ArtifactCandidate{
|
|
{Payload: []byte(`{"value":true}`)},
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
type integrationNormalizer struct{}
|
|
|
|
type integrationMerger struct{}
|
|
|
|
func (merger integrationMerger) Key() string {
|
|
return "merge"
|
|
}
|
|
|
|
func (merger integrationMerger) 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
|
|
}
|
|
|
|
func (normalizer integrationNormalizer) Key() string {
|
|
return "normalize"
|
|
}
|
|
|
|
func (normalizer integrationNormalizer) ReferenceSlots() []contracts.ReferenceSlot {
|
|
return nil
|
|
}
|
|
|
|
func (normalizer integrationNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
|
|
return contracts.NormalizeResult{Candidates: req.Candidates}, nil
|
|
}
|
|
|
|
type integrationOutput struct{}
|
|
|
|
func (output integrationOutput) Key() string {
|
|
return "output"
|
|
}
|
|
|
|
func (output integrationOutput) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
|
|
return contracts.OutputResult{
|
|
Files: []contracts.OutputFile{
|
|
{Name: "output.json", ContentType: "application/json", Bytes: []byte(`{}`)},
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
type integrationValidator struct {
|
|
name string
|
|
approve bool
|
|
}
|
|
|
|
func (validator integrationValidator) Name() string {
|
|
return validator.name
|
|
}
|
|
|
|
func (validator integrationValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
|
|
decisions := make([]contracts.ValidationDecision, 0, len(req.Candidates))
|
|
for _, candidate := range req.Candidates {
|
|
if validator.approve {
|
|
decisions = append(decisions, validate.Approved(candidate.Index))
|
|
} else {
|
|
decisions = append(decisions, validate.Rejected(candidate.Index, "invalid", "not accepted"))
|
|
}
|
|
}
|
|
return contracts.ValidationResult{
|
|
ValidatorName: validator.name,
|
|
Decisions: decisions,
|
|
}, nil
|
|
}
|
|
|
|
func integrationPipeline() ResolvedPipeline {
|
|
return ResolvedPipeline{
|
|
ID: "pipeline-1",
|
|
Digest: "sha256:pipeline",
|
|
Input: Binding("input"),
|
|
Chunk: Binding("chunk"),
|
|
ArtifactLanes: []ResolvedArtifactLane{
|
|
{
|
|
ID: "first",
|
|
Extract: Binding("extract-first"),
|
|
Merge: Binding("merge"),
|
|
Normalize: Binding("normalize"),
|
|
},
|
|
{
|
|
ID: "second",
|
|
Extract: Binding("extract-second"),
|
|
Merge: Binding("merge"),
|
|
Normalize: Binding("normalize"),
|
|
},
|
|
},
|
|
Output: Binding("output"),
|
|
}
|
|
}
|
|
|
|
func integrationSourceDocument() *source.SourceDocument {
|
|
return &source.SourceDocument{
|
|
ID: "source-1",
|
|
Kind: "document",
|
|
Format: "text/plain",
|
|
Digest: "sha256:abc123",
|
|
Units: []source.SourceUnit{
|
|
{ID: "u1", Kind: "unit", Text: "Source unit."},
|
|
},
|
|
}
|
|
}
|
|
|
|
func artifactKeys(approved []artifacts.Artifact) []string {
|
|
keys := make([]string, 0, len(approved))
|
|
for _, artifact := range approved {
|
|
keys = append(keys, artifact.ExtractorKey)
|
|
}
|
|
return keys
|
|
}
|
|
|
|
func rejectedKeys(rejected []artifacts.RejectedArtifact) []string {
|
|
keys := make([]string, 0, len(rejected))
|
|
for _, artifact := range rejected {
|
|
keys = append(keys, artifact.Candidate.ExtractorKey)
|
|
}
|
|
return keys
|
|
}
|