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")) }