342 lines
13 KiB
Go
342 lines
13 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"path/filepath"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
|
)
|
|
|
|
func TestDecideStageExecution(t *testing.T) {
|
|
statuses := []manifest.StageStatus{"", manifest.StatusSucceeded, manifest.StatusFailed, manifest.StatusStale, manifest.StatusSkipped}
|
|
validations := []struct {
|
|
name string
|
|
value *stage.ResumeValidation
|
|
}{
|
|
{name: "none"},
|
|
{name: "resumable", value: resumeValidation(stage.Resumable())},
|
|
{name: "not resumable", value: resumeValidation(stage.NonResumable("output changed"))},
|
|
}
|
|
for _, status := range statuses {
|
|
for _, force := range []bool{false, true} {
|
|
for _, validation := range validations {
|
|
name := fmt.Sprintf("status=%s/force=%t/resume=%s", status, force, validation.name)
|
|
t.Run(name, func(t *testing.T) {
|
|
got := decideStageExecution(status, force, validation.value)
|
|
want := preExecutionDecision{action: stageActionRun}
|
|
switch {
|
|
case force:
|
|
want.invalidationReason = staleReasonForcedReplacement
|
|
want.invalidateWhenStarted = true
|
|
case status == manifest.StatusSucceeded && validation.value != nil && !validation.value.Resumable:
|
|
want.markSelectedStale = true
|
|
want.selectedStaleReason = "output changed"
|
|
want.invalidationReason = staleReasonNotResumable
|
|
case status == manifest.StatusSucceeded:
|
|
want.action = stageActionSkip
|
|
}
|
|
if !reflect.DeepEqual(got, want) {
|
|
t.Fatalf("decision = %#v, want %#v", got, want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func resumeValidation(value stage.ResumeValidation) *stage.ResumeValidation { return &value }
|
|
|
|
func TestPlannedStageModelMatchesStageExecution(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
seed func(*manifest.Manifest, time.Time)
|
|
force bool
|
|
resume *stage.ResumeValidation
|
|
result *stage.StageResult
|
|
wantAction stageAction
|
|
wantStatus manifest.StageStatus
|
|
}{
|
|
{
|
|
name: "ordinary skip",
|
|
seed: func(m *manifest.Manifest, at time.Time) {
|
|
m.MarkStageSucceeded("extract", at, nil)
|
|
m.MarkStageSucceeded("analyze", at, nil)
|
|
},
|
|
wantAction: stageActionSkip,
|
|
wantStatus: manifest.StatusSucceeded,
|
|
},
|
|
{
|
|
name: "non-resumable success",
|
|
seed: func(m *manifest.Manifest, at time.Time) {
|
|
m.MarkStageSucceeded("extract", at, nil)
|
|
m.MarkStageSucceeded("analyze", at, nil)
|
|
},
|
|
resume: resumeValidation(stage.NonResumable("output changed")),
|
|
result: &stage.StageResult{},
|
|
wantAction: stageActionRun,
|
|
wantStatus: manifest.StatusStale,
|
|
},
|
|
{
|
|
name: "forced replacement",
|
|
seed: func(m *manifest.Manifest, at time.Time) {
|
|
m.MarkStageSucceeded("extract", at, nil)
|
|
m.MarkStageSucceeded("analyze", at, nil)
|
|
},
|
|
force: true,
|
|
result: &stage.StageResult{},
|
|
wantAction: stageActionRun,
|
|
wantStatus: manifest.StatusStale,
|
|
},
|
|
{
|
|
name: "unchanged self-skip",
|
|
seed: func(m *manifest.Manifest, at time.Time) {
|
|
m.MarkStageSkipped("extract", at, "notarius_disabled")
|
|
m.MarkStageSucceeded("analyze", at, nil)
|
|
},
|
|
result: &stage.StageResult{
|
|
Disposition: stage.StageDispositionSkipped,
|
|
SkipReason: "notarius_disabled",
|
|
},
|
|
wantAction: stageActionRun,
|
|
wantStatus: manifest.StatusSucceeded,
|
|
},
|
|
{
|
|
name: "changed self-skip",
|
|
seed: func(m *manifest.Manifest, at time.Time) {
|
|
m.MarkStageSkipped("extract", at, "integration_disabled")
|
|
m.MarkStageSucceeded("analyze", at, nil)
|
|
},
|
|
result: &stage.StageResult{
|
|
Disposition: stage.StageDispositionSkipped,
|
|
SkipReason: "notarius_disabled",
|
|
},
|
|
wantAction: stageActionRun,
|
|
wantStatus: manifest.StatusStale,
|
|
},
|
|
{
|
|
name: "success replacing non-success",
|
|
seed: func(m *manifest.Manifest, at time.Time) {
|
|
m.MarkStageFailed("extract", at, "failed")
|
|
m.MarkStageSucceeded("analyze", at, nil)
|
|
},
|
|
result: &stage.StageResult{},
|
|
wantAction: stageActionRun,
|
|
wantStatus: manifest.StatusStale,
|
|
},
|
|
}
|
|
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
at := time.Date(2026, time.May, 3, 12, 0, 0, 0, time.UTC)
|
|
cfg := testConfig(t)
|
|
workDir := filepath.Join(t.TempDir(), "run")
|
|
cfg.Pipeline.Notarius = &config.NotariusConfig{Enabled: test.result == nil || test.result.Disposition != stage.StageDispositionSkipped}
|
|
planned := manifest.New("2026-05-03", at)
|
|
executed := manifest.New("2026-05-03", at)
|
|
planned.Campaign = "sample-campaign"
|
|
executed.Campaign = "sample-campaign"
|
|
test.seed(planned, at)
|
|
test.seed(executed, at)
|
|
|
|
plannedAction := decideStageExecution(recordedStageStatus(planned, "extract"), test.force, test.resume)
|
|
if _, err := applyPreExecutionDecision(planned, "extract", plannedAction, at.Add(time.Second)); err != nil {
|
|
t.Fatalf("apply planning decision: %v", err)
|
|
}
|
|
selected := transitionParityStage{validation: stage.Resumable(), result: test.result}
|
|
if test.resume != nil {
|
|
selected.validation = *test.resume
|
|
}
|
|
if plannedAction.action == stageActionRun {
|
|
if err := modelPlannedStageRun(planned, selected, cfg, plannedAction, nil); err != nil {
|
|
t.Fatalf("modelPlannedStageRun() error = %v", err)
|
|
}
|
|
}
|
|
|
|
runtime := &invocationRuntime{
|
|
ctx: context.Background(),
|
|
cfg: cfg,
|
|
opts: RunOptions{Force: test.force},
|
|
identity: invocationIdentity{RunID: "20260503T120000Z-a1b2c3d4", LocalWorkDir: workDir},
|
|
env: &stage.Env{Config: cfg, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))},
|
|
manifestStore: &manifest.LocalStore{},
|
|
manifestPath: filepath.Join(t.TempDir(), "manifest.json"),
|
|
manifest: executed,
|
|
dependencies: invocationDependencies{
|
|
now: func() time.Time { return at.Add(time.Second) },
|
|
}.normalized(),
|
|
}
|
|
executedOutcome, err := executeSelectedStage(runtime, selected)
|
|
if err != nil {
|
|
t.Fatalf("executeSelectedStage() error = %v", err)
|
|
}
|
|
executedAction := stageActionRun
|
|
if executedOutcome.skipped && !executedOutcome.executed {
|
|
executedAction = stageActionSkip
|
|
}
|
|
if plannedAction.action != test.wantAction || executedAction != test.wantAction {
|
|
t.Fatalf("actions = planned %q, executed %q; want %q", plannedAction.action, executedAction, test.wantAction)
|
|
}
|
|
if got := planned.Stages["analyze"].Status; got != test.wantStatus {
|
|
t.Fatalf("planned dependent status = %q, want %q", got, test.wantStatus)
|
|
}
|
|
if got := executed.Stages["analyze"].Status; got != test.wantStatus {
|
|
t.Fatalf("executed dependent status = %q, want %q", got, test.wantStatus)
|
|
}
|
|
if got, want := executed.Stages["extract"].Status, planned.Stages["extract"].Status; got != want {
|
|
t.Fatalf("selected stage status = %q, want planned status %q", got, want)
|
|
}
|
|
plannedError := planned.Stages["analyze"].Error
|
|
executedError := executed.Stages["analyze"].Error
|
|
if (plannedError == nil) != (executedError == nil) ||
|
|
(plannedError != nil && (plannedError.Code != executedError.Code || plannedError.Message != executedError.Message)) {
|
|
t.Fatalf("dependent errors differ: planned %#v, executed %#v", plannedError, executedError)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
type transitionParityStage struct {
|
|
validation stage.ResumeValidation
|
|
result *stage.StageResult
|
|
}
|
|
|
|
func (transitionParityStage) Name() string { return "extract" }
|
|
|
|
func (s transitionParityStage) Run(context.Context, *stage.Env, *manifest.Manifest) (*stage.StageResult, error) {
|
|
return s.result, nil
|
|
}
|
|
|
|
func (s transitionParityStage) ValidateResume(context.Context, *stage.Env, *manifest.Manifest) (stage.ResumeValidation, error) {
|
|
return s.validation, nil
|
|
}
|
|
|
|
func TestInvalidationDependents(t *testing.T) {
|
|
tests := []struct {
|
|
stage string
|
|
want []string
|
|
}{
|
|
{"prepare", []string{"transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}},
|
|
{"transcribe", []string{"merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}},
|
|
{"merge", []string{"polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}},
|
|
{"polish", []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}},
|
|
{"normalize", []string{"trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}},
|
|
{"trim", []string{"render", "extract", "analyze", "publish", "notify", "cleanup"}},
|
|
{"render", []string{"analyze", "publish", "notify", "cleanup"}},
|
|
{"extract", []string{"analyze", "publish", "notify", "cleanup"}},
|
|
{"analyze", []string{"publish", "notify", "cleanup"}},
|
|
{"publish", []string{"notify", "cleanup"}},
|
|
{"notify", []string{"cleanup"}},
|
|
{"cleanup", []string{}},
|
|
}
|
|
for _, test := range tests {
|
|
t.Run(test.stage, func(t *testing.T) {
|
|
got, err := dependentStageNames(test.stage)
|
|
if err != nil {
|
|
t.Fatalf("dependentStageNames(%q) error = %v", test.stage, err)
|
|
}
|
|
if !reflect.DeepEqual(got, test.want) {
|
|
t.Fatalf("dependentStageNames(%q) = %#v, want %#v", test.stage, got, test.want)
|
|
}
|
|
})
|
|
}
|
|
if _, err := dependentStageNames("unknown"); err == nil || !strings.Contains(err.Error(), "unknown stage") {
|
|
t.Fatalf("dependentStageNames(unknown) error = %v, want unknown-stage error", err)
|
|
}
|
|
}
|
|
|
|
func TestInvalidationRelationRejectsInvalidInventory(t *testing.T) {
|
|
canonical := []stage.Stage{
|
|
invalidationTestStage("one"),
|
|
invalidationTestStage("two"),
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
registry []stage.Stage
|
|
edges map[string][]string
|
|
want string
|
|
}{
|
|
{name: "duplicate registry name", registry: append(canonical, invalidationTestStage("one")), edges: map[string][]string{"one": {"two"}, "two": {}}, want: "duplicate stage"},
|
|
{name: "unknown source", registry: canonical, edges: map[string][]string{"one": {"two"}, "two": {}, "three": {}}, want: "unknown stage"},
|
|
{name: "unknown target", registry: canonical, edges: map[string][]string{"one": {"three"}, "two": {}}, want: "unknown stage"},
|
|
{name: "missing classification", registry: canonical, edges: map[string][]string{"one": {"two"}}, want: "missing classification"},
|
|
{name: "cycle", registry: canonical, edges: map[string][]string{"one": {"two"}, "two": {"one"}}, want: "cycle"},
|
|
}
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
_, err := newInvalidationRelation(test.registry, test.edges)
|
|
if err == nil || !strings.Contains(err.Error(), test.want) {
|
|
t.Fatalf("newInvalidationRelation() error = %v, want %q", err, test.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestInvalidateDependentSucceededStagesWithReason(t *testing.T) {
|
|
now := time.Now().UTC()
|
|
m := manifest.New("2026-05-03", now)
|
|
for _, name := range canonicalStageNames() {
|
|
m.MarkStageSucceeded(name, now, nil)
|
|
}
|
|
m.MarkStageFailed("analyze", now, "analysis failed")
|
|
|
|
got, err := invalidateDependentSucceededStagesWithReason(m, "polish", now.Add(time.Second), staleReasonChangedResult)
|
|
if err != nil {
|
|
t.Fatalf("invalidateDependentSucceededStagesWithReason() error = %v", err)
|
|
}
|
|
want := []string{"normalize", "trim", "render", "extract", "publish", "notify", "cleanup"}
|
|
if !reflect.DeepEqual(got, want) {
|
|
t.Fatalf("invalidateDependentSucceededStagesWithReason() = %#v, want %#v", got, want)
|
|
}
|
|
for _, stageName := range want {
|
|
if m.Stages[stageName].Status != manifest.StatusStale {
|
|
t.Fatalf("%s status = %q, want stale", stageName, m.Stages[stageName].Status)
|
|
}
|
|
}
|
|
if m.Stages["analyze"].Status != manifest.StatusFailed {
|
|
t.Fatalf("analyze status = %q, want failed", m.Stages["analyze"].Status)
|
|
}
|
|
}
|
|
|
|
func TestRenderAndExtractInvalidationAreIndependent(t *testing.T) {
|
|
now := time.Now().UTC()
|
|
for _, upstream := range []string{"render", "extract"} {
|
|
t.Run(upstream, func(t *testing.T) {
|
|
m := manifest.New("2026-05-03", now)
|
|
for _, name := range canonicalStageNames() {
|
|
m.MarkStageSucceeded(name, now, nil)
|
|
}
|
|
got, err := invalidateDependentSucceededStagesWithReason(m, upstream, now.Add(time.Second), staleReasonForcedReplacement)
|
|
if err != nil {
|
|
t.Fatalf("invalidate dependents: %v", err)
|
|
}
|
|
want := []string{"analyze", "publish", "notify", "cleanup"}
|
|
if !reflect.DeepEqual(got, want) {
|
|
t.Fatalf("invalidated = %#v, want %#v", got, want)
|
|
}
|
|
sibling := "render"
|
|
if upstream == "render" {
|
|
sibling = "extract"
|
|
}
|
|
if m.Stages[sibling].Status != manifest.StatusSucceeded {
|
|
t.Fatalf("%s invalidated sibling %s: %#v", upstream, sibling, m.Stages[sibling])
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
type invalidationTestStage string
|
|
|
|
func (s invalidationTestStage) Name() string { return string(s) }
|
|
func (s invalidationTestStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
|
|
return &stage.StageResult{}, nil
|
|
}
|