diff --git a/internal/progress/tracker_test.go b/internal/progress/tracker_test.go new file mode 100644 index 0000000..d88a88f --- /dev/null +++ b/internal/progress/tracker_test.go @@ -0,0 +1,295 @@ +package progress_test + +import ( + "bytes" + "context" + "errors" + "log/slog" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "git.d-ma.be/mathias/tapir/internal/domain" + "git.d-ma.be/mathias/tapir/internal/progress" +) + +// fakeSink records every saved snapshot and every heartbeat. +type fakeSink struct { + mu sync.Mutex + saves [][]domain.ProgressStep + at []time.Time + beats int + err error +} + +func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, at time.Time) error { + f.mu.Lock() + defer f.mu.Unlock() + f.saves = append(f.saves, append([]domain.ProgressStep(nil), steps...)) + f.at = append(f.at, at) + return f.err +} + +func (f *fakeSink) Beat(context.Context) error { + f.mu.Lock() + defer f.mu.Unlock() + f.beats++ + return f.err +} + +func (f *fakeSink) last(t *testing.T) []domain.ProgressStep { + t.Helper() + f.mu.Lock() + defer f.mu.Unlock() + require.NotEmpty(t, f.saves, "nothing was saved") + return f.saves[len(f.saves)-1] +} + +func (f *fakeSink) saveCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.saves) +} + +func (f *fakeSink) beatCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.beats +} + +// clock is a hand-wound clock so durations and throttling are exact. +type clock struct{ t time.Time } + +func (c *clock) now() time.Time { return c.t } +func (c *clock) tick(d time.Duration) { c.t = c.t.Add(d) } +func newClock() *clock { return &clock{t: time.Date(2026, 10, 2, 8, 0, 0, 0, time.UTC)} } + +func plan() []progress.Planned { + return []progress.Planned{{Key: "audio", Label: "Preparing audio"}, {Key: "decide", Label: "Deciding"}, {Key: "save", Label: "Saving"}} +} + +func newTracker(t *testing.T) (*progress.Tracker, *fakeSink, *clock, *bytes.Buffer) { + t.Helper() + sink, clk, buf := &fakeSink{}, newClock(), &bytes.Buffer{} + log := slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + return progress.NewTracker(context.Background(), sink, log, plan(), progress.WithClock(clk.now), progress.WithMinSave(2*time.Second)), sink, clk, buf +} + +func stateOf(steps []domain.ProgressStep, key string) domain.StepState { + for _, s := range steps { + if s.Key == key { + return s.State + } + } + return "" +} + +func TestTracker_PlanIsShownUpFrontInOrderAllPending(t *testing.T) { + tr, sink, _, _ := newTracker(t) + + tr.Start() + + steps := sink.last(t) + require.Len(t, steps, 3) + require.Equal(t, []string{"audio", "decide", "save"}, []string{steps[0].Key, steps[1].Key, steps[2].Key}) + for _, s := range steps { + require.Equal(t, domain.StepPending, s.State, s.Key) + } + require.Equal(t, "Preparing audio", steps[0].Label) +} + +func TestTracker_BeginRunsAStepAndDoneRecordsItsDuration(t *testing.T) { + tr, sink, clk, _ := newTracker(t) + tr.Start() + + tr.Begin("audio", 0) + require.Equal(t, domain.StepRunning, stateOf(sink.last(t), "audio")) + clk.tick(90 * time.Second) + tr.Done("audio") + + s := sink.last(t)[0] + require.Equal(t, domain.StepDone, s.State) + require.Equal(t, 90*time.Second, s.FinishedAt.Sub(s.StartedAt), "the finished step must carry how long it took") +} + +func TestTracker_AdvanceCountsAreSavedAndThrottled(t *testing.T) { + tr, sink, clk, _ := newTracker(t) + tr.Start() + tr.Begin("decide", 282) + base := sink.saveCount() + + tr.Advance("decide", 1) + clk.tick(500 * time.Millisecond) + tr.Advance("decide", 2) + clk.tick(500 * time.Millisecond) + tr.Advance("decide", 3) + + require.Equal(t, base+1, sink.saveCount(), "advances inside the minimum interval must not each hit the database") + clk.tick(2 * time.Second) + tr.Advance("decide", 4) + require.Equal(t, base+2, sink.saveCount()) + got := sink.last(t)[1] + require.Equal(t, 4, got.Done) + require.Equal(t, 282, got.Total) +} + +func TestTracker_DoneFlushesTheFinalCountEvenInsideTheThrottle(t *testing.T) { + tr, sink, _, _ := newTracker(t) + tr.Start() + tr.Begin("decide", 10) + tr.Advance("decide", 9) + + tr.Advance("decide", 10) + tr.Done("decide") + + got := sink.last(t)[1] + require.Equal(t, domain.StepDone, got.State) + require.Equal(t, 10, got.Done, "a finished step must show its full count, not the last throttled one") +} + +func TestTracker_FinishSkipsWhatNeverRanAndClosesWhatDid(t *testing.T) { + tr, sink, _, _ := newTracker(t) + tr.Start() + tr.Begin("audio", 0) + tr.Done("audio") + tr.Begin("decide", 5) + + tr.Finish() + + steps := sink.last(t) + require.Equal(t, domain.StepDone, stateOf(steps, "audio")) + require.Equal(t, domain.StepDone, stateOf(steps, "decide"), "a step still running when the job completes is closed") + require.Equal(t, domain.StepSkipped, stateOf(steps, "save"), "a step that never ran is skipped, not left pending") +} + +func TestTracker_FailMarksTheRunningStepFailedAndLeavesTheRestPending(t *testing.T) { + tr, sink, _, _ := newTracker(t) + tr.Start() + tr.Begin("audio", 0) + tr.Done("audio") + tr.Begin("decide", 5) + + tr.Fail() + + steps := sink.last(t) + require.Equal(t, domain.StepDone, stateOf(steps, "audio")) + require.Equal(t, domain.StepFailed, stateOf(steps, "decide")) + require.Equal(t, domain.StepPending, stateOf(steps, "save"), "what never started stays pending: it shows where the job stopped") +} + +func TestTracker_AnUnknownKeyIsIgnored(t *testing.T) { + tr, sink, _, _ := newTracker(t) + tr.Start() + before := sink.saveCount() + + require.NotPanics(t, func() { + tr.Begin("no-such-step", 0) + tr.Advance("no-such-step", 1) + tr.Done("no-such-step") + }) + require.Equal(t, before, sink.saveCount()) +} + +func TestTracker_ASinkFailureIsLoggedAndNeverStopsTheJob(t *testing.T) { + tr, sink, _, buf := newTracker(t) + sink.err = errors.New("db down") + + require.NotPanics(t, func() { + tr.Start() + tr.Begin("audio", 0) + tr.Done("audio") + }) + require.Contains(t, buf.String(), "level=WARN", "a lost progress write is worth a WARN") + require.Contains(t, buf.String(), "db down") +} + +func TestTracker_EveryTransitionIsLoggedByTheSameCall(t *testing.T) { + tr, _, clk, buf := newTracker(t) + tr.Start() + tr.Begin("audio", 0) + clk.tick(3 * time.Second) + tr.Done("audio") + + out := buf.String() + require.Contains(t, out, "step=audio") + require.Contains(t, out, "started") + require.Contains(t, out, "finished") + require.Contains(t, out, "elapsed=3s") +} + +func TestTracker_AdvanceIsLoggedAtMostOncePerMinute(t *testing.T) { + tr, _, clk, buf := newTracker(t) + tr.Start() + tr.Begin("decide", 100) + + for i := 1; i <= 30; i++ { + clk.tick(1 * time.Second) + tr.Advance("decide", i) + } + require.LessOrEqual(t, strings.Count(buf.String(), "step progress"), 1, + "30 seconds of progress is at most one log line, not thirty") + clk.tick(61 * time.Second) + tr.Advance("decide", 31) + require.Equal(t, 2, strings.Count(buf.String(), "step progress")) + require.Contains(t, buf.String(), "done=31") + require.Contains(t, buf.String(), "total=100") +} + +func TestTracker_HeartbeatBeatsOnItsOwnUntilStopped(t *testing.T) { + tr, sink, _, _ := newTracker(t) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + + go func() { + defer close(done) + tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour) + }() + + require.Eventually(t, func() bool { return sink.beatCount() >= 3 }, 2*time.Second, 5*time.Millisecond, + "a heartbeat that does not depend on any engine call is the point") + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Heartbeat did not stop when its context ended") + } +} + +func TestTracker_HeartbeatSaysWhichStepIsStillRunningAndForHowLong(t *testing.T) { + tr, _, clk, buf := newTracker(t) + tr.Start() + tr.Begin("decide", 100) + clk.tick(7 * time.Minute) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go tr.Heartbeat(ctx, time.Hour, 5*time.Millisecond) + + require.Eventually(t, func() bool { return strings.Contains(buf.String(), "still running") }, 2*time.Second, 5*time.Millisecond) + require.Contains(t, buf.String(), "step=decide") + require.Contains(t, buf.String(), "elapsed=7m0s") +} + +func TestFrom_WithoutAReporterReturnsOneThatDoesNothing(t *testing.T) { + r := progress.From(context.Background()) + + require.NotNil(t, r) + require.NotPanics(t, func() { + r.Begin("x", 3) + r.Advance("x", 1) + r.Done("x") + }) +} + +func TestWithAndFrom_ACarriedReporterIsTheOneReturned(t *testing.T) { + tr, sink, _, _ := newTracker(t) + tr.Start() + ctx := progress.With(context.Background(), tr) + + progress.From(ctx).Begin("audio", 0) + + require.Equal(t, domain.StepRunning, stateOf(sink.last(t), "audio")) +}