273 lines
11 KiB
Go
273 lines
11 KiB
Go
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
|
)
|
|
|
|
type observedNotesCodec struct {
|
|
candidateValues []codecNotes
|
|
finalValues []codecNotes
|
|
candidateError string
|
|
finalError string
|
|
}
|
|
|
|
func (*observedNotesCodec) Kind() contracts.ArtifactKind { return "test/notes" }
|
|
|
|
func (*observedNotesCodec) Schema() contracts.ArtifactSchema { return notesCodec().schema }
|
|
|
|
func (*observedNotesCodec) MediaType() string { return "application/json" }
|
|
|
|
func (c *observedNotesCodec) EncodeCandidate(value codecNotes) ([]byte, error) {
|
|
c.candidateValues = append(c.candidateValues, value)
|
|
if c.candidateError != "" && firstNote(value) == c.candidateError {
|
|
return nil, errors.New("candidate encoding failed")
|
|
}
|
|
return json.Marshal(value)
|
|
}
|
|
|
|
func (c *observedNotesCodec) Encode(value codecNotes) ([]byte, error) {
|
|
c.finalValues = append(c.finalValues, value)
|
|
if c.finalError != "" && firstNote(value) == c.finalError {
|
|
return nil, errors.New("final encoding failed")
|
|
}
|
|
if firstNote(value) == "invalid" {
|
|
return nil, errors.New("invalid note is not a final artifact")
|
|
}
|
|
return json.Marshal(value)
|
|
}
|
|
|
|
func (*observedNotesCodec) Decode(content []byte) (codecNotes, error) {
|
|
var value codecNotes
|
|
return value, json.Unmarshal(content, &value)
|
|
}
|
|
|
|
func (c *observedNotesCodec) DecodeCandidate(content []byte) (codecNotes, error) {
|
|
return c.Decode(content)
|
|
}
|
|
|
|
func firstNote(value codecNotes) string {
|
|
if len(value.Items) == 0 {
|
|
return ""
|
|
}
|
|
return value.Items[0]
|
|
}
|
|
|
|
func (c *observedNotesCodec) candidateCalls(value string) int {
|
|
return matchingNotes(c.candidateValues, value)
|
|
}
|
|
|
|
func (c *observedNotesCodec) finalCalls(value string) int {
|
|
return matchingNotes(c.finalValues, value)
|
|
}
|
|
|
|
func matchingNotes(values []codecNotes, value string) int {
|
|
count := 0
|
|
for _, candidate := range values {
|
|
if firstNote(candidate) == value {
|
|
count++
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
|
|
type candidateCheckpointRecorder struct {
|
|
CheckpointRecorder
|
|
mergeSucceeded int
|
|
normalizeSucceeded int
|
|
mergeOutput CheckpointArtifact
|
|
normalizeOutput CheckpointArtifact
|
|
}
|
|
|
|
func (r *candidateCheckpointRecorder) MergeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact, _ []contracts.Warning) error {
|
|
r.mergeSucceeded++
|
|
r.mergeOutput = cloneCheckpointArtifact(output)
|
|
return nil
|
|
}
|
|
|
|
func (r *candidateCheckpointRecorder) NormalizeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact, _ []contracts.Warning) error {
|
|
r.normalizeSucceeded++
|
|
r.normalizeOutput = cloneCheckpointArtifact(output)
|
|
return nil
|
|
}
|
|
|
|
func installObservedNotesCodec(t *testing.T, prepared *PreparedPipeline, codec *observedNotesCodec) {
|
|
t.Helper()
|
|
registry := NewArtifactCodecRegistry()
|
|
if err := RegisterArtifactCodec(registry, codec); err != nil {
|
|
t.Fatalf("RegisterArtifactCodec() error = %v", err)
|
|
}
|
|
entry, _, err := registry.entry("test/notes")
|
|
if err != nil {
|
|
t.Fatalf("codec entry error = %v", err)
|
|
}
|
|
prepared.Steps[0].lanes[0].typed.codec = entry
|
|
}
|
|
|
|
func configureCandidateOperation(prepared *PreparedPipeline, target ModuleStage, value codecNotes) {
|
|
lane := &prepared.Steps[0].lanes[0]
|
|
switch target {
|
|
case StageMerge:
|
|
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
|
|
return erasedTypedResult{Value: value}, nil
|
|
}
|
|
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
|
|
return erasedTypedResult{Value: codecNotes{Items: []string{"normalized-other"}}}, nil
|
|
}
|
|
case StageNormalize:
|
|
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
|
|
return erasedTypedResult{Value: codecNotes{Items: []string{"merged-other"}}}, nil
|
|
}
|
|
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
|
|
return erasedTypedResult{Value: value}, nil
|
|
}
|
|
}
|
|
}
|
|
|
|
func setCandidateValidator(prepared *PreparedPipeline, target ModuleStage, approved bool) {
|
|
validator := preparedValidator{
|
|
resolved: ResolvedValidator{Binding: Binding("candidate-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
|
|
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
|
|
return contracts.ValidationResult{Approved: approved, ReasonCode: "candidate_rejected", Message: "candidate rejected by validator"}, nil
|
|
},
|
|
}
|
|
switch target {
|
|
case StageMerge:
|
|
prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{validator}
|
|
case StageNormalize:
|
|
prepared.Steps[0].lanes[0].normalizeValidators.validators = []preparedValidator{validator}
|
|
}
|
|
}
|
|
|
|
func TestRunnerRejectsCandidatesBeforeFinalEncoding(t *testing.T) {
|
|
for _, target := range []ModuleStage{StageMerge, StageNormalize} {
|
|
t.Run(string(target), func(t *testing.T) {
|
|
prepared := preparedAttemptDebugPipeline(t)
|
|
codec := &observedNotesCodec{}
|
|
installObservedNotesCodec(t, prepared, codec)
|
|
configureCandidateOperation(prepared, target, codecNotes{Items: []string{"invalid"}})
|
|
setCandidateValidator(prepared, target, false)
|
|
recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
|
|
|
|
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder})
|
|
if err != nil {
|
|
t.Fatalf("Run() error = %v, want validator rejection", err)
|
|
}
|
|
if len(output.Rejected) != 1 || output.Rejected[0].Stage != string(target) || output.Rejected[0].ReasonCode != "candidate_rejected" {
|
|
t.Fatalf("rejected outputs = %#v, want %s validator rejection", output.Rejected, target)
|
|
}
|
|
if codec.candidateCalls("invalid") != 1 || codec.finalCalls("invalid") != 0 {
|
|
t.Fatalf("invalid candidate calls = candidate %d, final %d; want 1, 0", codec.candidateCalls("invalid"), codec.finalCalls("invalid"))
|
|
}
|
|
if target == StageMerge && recorder.mergeSucceeded != 0 {
|
|
t.Fatalf("merge checkpoints = %d, want none", recorder.mergeSucceeded)
|
|
}
|
|
if target == StageNormalize && recorder.normalizeSucceeded != 0 {
|
|
t.Fatalf("normalize checkpoints = %d, want none", recorder.normalizeSucceeded)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRunnerFinalEncodesAcceptedCandidatesOnce(t *testing.T) {
|
|
for _, target := range []ModuleStage{StageMerge, StageNormalize} {
|
|
t.Run(string(target), func(t *testing.T) {
|
|
prepared := preparedAttemptDebugPipeline(t)
|
|
codec := &observedNotesCodec{}
|
|
installObservedNotesCodec(t, prepared, codec)
|
|
value := "accepted-" + string(target)
|
|
configureCandidateOperation(prepared, target, codecNotes{Items: []string{value}})
|
|
setCandidateValidator(prepared, target, true)
|
|
recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
|
|
loader := &extractResultLoader{CheckpointLoader: NoopCheckpointLoader(), decision: CheckpointDecision{Reason: "not found"}}
|
|
|
|
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Checkpoint: loader})
|
|
if err != nil {
|
|
t.Fatalf("Run() error = %v, want nil", err)
|
|
}
|
|
if codec.candidateCalls(value) != 1 || codec.finalCalls(value) != 1 {
|
|
t.Fatalf("accepted candidate calls = candidate %d, final %d; want 1, 1", codec.candidateCalls(value), codec.finalCalls(value))
|
|
}
|
|
if target == StageMerge && recorder.mergeSucceeded != 1 {
|
|
t.Fatalf("merge checkpoints = %d, want one", recorder.mergeSucceeded)
|
|
}
|
|
if target == StageNormalize && recorder.normalizeSucceeded != 1 {
|
|
t.Fatalf("normalize checkpoints = %d, want one", recorder.normalizeSucceeded)
|
|
}
|
|
checkpoint := recorder.mergeOutput
|
|
if target == StageNormalize {
|
|
checkpoint = recorder.normalizeOutput
|
|
}
|
|
var stored codecNotes
|
|
if err := json.Unmarshal(checkpoint.Artifact.Content, &stored); err != nil || !reflect.DeepEqual(stored, codecNotes{Items: []string{value}}) {
|
|
t.Fatalf("checkpoint content = %s, %v; want accepted value %q", checkpoint.Artifact.Content, err, value)
|
|
}
|
|
gotStages := make([]string, len(output.CheckpointEvents))
|
|
for i, event := range output.CheckpointEvents {
|
|
gotStages[i] = event.Stage
|
|
}
|
|
wantStages := []string{"source", string(StageExtract), string(StageMerge), string(StageNormalize)}
|
|
if !reflect.DeepEqual(gotStages, wantStages) {
|
|
t.Fatalf("checkpoint event stages = %#v, want %#v", gotStages, wantStages)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRunnerRecordsCandidateAndFinalEncodingFailures(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
target ModuleStage
|
|
value string
|
|
candidate bool
|
|
wantError string
|
|
}{
|
|
{name: "merge candidate", target: StageMerge, value: "merge-candidate-failure", candidate: true, wantError: "serialize merge candidate"},
|
|
{name: "normalize candidate", target: StageNormalize, value: "normalize-candidate-failure", candidate: true, wantError: "serialize normalize candidate"},
|
|
{name: "merge final", target: StageMerge, value: "merge-final-failure", wantError: "serialize accepted merge output"},
|
|
{name: "normalize final", target: StageNormalize, value: "normalize-final-failure", wantError: "serialize accepted normalize output"},
|
|
}
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
prepared := preparedAttemptDebugPipeline(t)
|
|
codec := &observedNotesCodec{}
|
|
if tc.candidate {
|
|
codec.candidateError = tc.value
|
|
} else {
|
|
codec.finalError = tc.value
|
|
}
|
|
installObservedNotesCodec(t, prepared, codec)
|
|
configureCandidateOperation(prepared, tc.target, codecNotes{Items: []string{tc.value}})
|
|
setCandidateValidator(prepared, tc.target, true)
|
|
recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
|
|
debug := newCapturedDebugRecorder()
|
|
|
|
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Debug: debug})
|
|
if err == nil || !strings.Contains(err.Error(), tc.wantError) {
|
|
t.Fatalf("Run() error = %v, want %q", err, tc.wantError)
|
|
}
|
|
attemptPath := fmt.Sprintf("%s/notes/attempt-01.json", tc.target)
|
|
envelope := debug.envelope(t, attemptPath)
|
|
if !strings.Contains(envelope.Error, tc.wantError) {
|
|
t.Fatalf("attempt error = %q, want %q", envelope.Error, tc.wantError)
|
|
}
|
|
if tc.candidate && codec.finalCalls(tc.value) != 0 {
|
|
t.Fatalf("final encode calls = %d, want none after candidate failure", codec.finalCalls(tc.value))
|
|
}
|
|
if tc.target == StageMerge && recorder.mergeSucceeded != 0 {
|
|
t.Fatalf("merge checkpoints = %d, want none", recorder.mergeSucceeded)
|
|
}
|
|
if tc.target == StageNormalize && recorder.normalizeSucceeded != 0 {
|
|
t.Fatalf("normalize checkpoints = %d, want none", recorder.normalizeSucceeded)
|
|
}
|
|
})
|
|
}
|
|
}
|