test(transcribe): the ensemble declares and reports its phases; a failed engine is shown failed, a shutdown is not (red)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,162 @@
|
|||||||
|
package transcribe_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/adapters/transcribe"
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/progress"
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/reconcile"
|
||||||
|
)
|
||||||
|
|
||||||
|
// What the Ensemble reports to the page and the log (ADR-035). The reporter here is a
|
||||||
|
// recording double carried in the context, exactly as the worker's Tracker is.
|
||||||
|
|
||||||
|
type recReporter struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
events []string
|
||||||
|
notes []map[string]any
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *recReporter) add(e string) {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
r.events = append(r.events, e)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *recReporter) Begin(key string, _ int) { r.add("begin:" + key) }
|
||||||
|
func (r *recReporter) Advance(key string, _ int) { r.add("advance:" + key) }
|
||||||
|
func (r *recReporter) Done(key string) { r.add("done:" + key) }
|
||||||
|
func (r *recReporter) Fail(key string, _ error) { r.add("fail:" + key) }
|
||||||
|
func (r *recReporter) Note(msg string, kv ...any) {
|
||||||
|
m := map[string]any{"msg": msg}
|
||||||
|
for i := 0; i+1 < len(kv); i += 2 {
|
||||||
|
m[kv[i].(string)] = kv[i+1]
|
||||||
|
}
|
||||||
|
r.mu.Lock()
|
||||||
|
r.notes = append(r.notes, m)
|
||||||
|
r.mu.Unlock()
|
||||||
|
r.add("note:" + msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *recReporter) joined() string {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
return strings.Join(r.events, ",")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsemblePlan_DeclaresBothEnginesAndTheReconcileStepsByName(t *testing.T) {
|
||||||
|
r := newEnsemble("a", "a", nil, nil, false, &verdictAdj{})
|
||||||
|
|
||||||
|
plan := r.e.Plan()
|
||||||
|
|
||||||
|
require.Len(t, plan, 4)
|
||||||
|
assert.Equal(t, []string{"engine_a", "engine_b", "compare", "decide"}, []string{plan[0].Key, plan[1].Key, plan[2].Key, plan[3].Key})
|
||||||
|
assert.Contains(t, plan[0].Label, "iguana/pianissimo", "the page says which engine is running")
|
||||||
|
assert.Contains(t, plan[1].Label, "iguana/kb-whisper")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsemblePlan_TheCloudFallbackIsAStepOnlyWhenOneIsConfigured(t *testing.T) {
|
||||||
|
with := newEnsemble("a", "a", nil, nil, true, &verdictAdj{}).e.Plan()
|
||||||
|
without := newEnsemble("a", "a", nil, nil, false, &verdictAdj{}).e.Plan()
|
||||||
|
|
||||||
|
assert.Equal(t, "fallback", with[len(with)-1].Key)
|
||||||
|
assert.Len(t, without, 4, "no fallback configured: the page must not promise one")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsembleProgress_BothEnginesAreReportedInTheOrderTheyRun(t *testing.T) {
|
||||||
|
rep := &recReporter{}
|
||||||
|
r := newEnsemble("Totalt 1015 kronor idag.", "Totalt 1050 kronor idag.", nil, nil, false, &verdictAdj{v: reconcile.VerdictAlt})
|
||||||
|
|
||||||
|
_, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
got := rep.joined()
|
||||||
|
a, aDone := strings.Index(got, "begin:engine_a"), strings.Index(got, "done:engine_a")
|
||||||
|
b, bDone := strings.Index(got, "begin:engine_b"), strings.Index(got, "done:engine_b")
|
||||||
|
require.True(t, a >= 0 && aDone > a && b > aDone && bDone > b, "events out of order: %s", got)
|
||||||
|
assert.NotContains(t, got, "fail:")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsembleProgress_AnEngineThatFailsIsShownFailedAndTheOtherStillRuns(t *testing.T) {
|
||||||
|
rep := &recReporter{}
|
||||||
|
r := newEnsemble("text", "text", errors.New("pianissimo down"), nil, false, &verdictAdj{})
|
||||||
|
|
||||||
|
got, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")
|
||||||
|
|
||||||
|
require.NoError(t, err, "one engine down is a degraded result, not a failed job")
|
||||||
|
assert.Equal(t, domain.ReconcileSingle, got.Reconciliation.Status)
|
||||||
|
ev := rep.joined()
|
||||||
|
assert.Contains(t, ev, "fail:engine_a")
|
||||||
|
assert.Contains(t, ev, "done:engine_b")
|
||||||
|
assert.NotContains(t, ev, "begin:compare", "nothing to compare with one engine")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsembleProgress_TheFallbackIsReportedOnlyWhenBothEnginesFailed(t *testing.T) {
|
||||||
|
rep := &recReporter{}
|
||||||
|
r := newEnsemble("t", "t", errors.New("a down"), errors.New("b down"), true, &verdictAdj{})
|
||||||
|
|
||||||
|
_, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
ev := rep.joined()
|
||||||
|
assert.Contains(t, ev, "fail:engine_a")
|
||||||
|
assert.Contains(t, ev, "fail:engine_b")
|
||||||
|
assert.Contains(t, ev, "begin:fallback")
|
||||||
|
assert.Contains(t, ev, "done:fallback")
|
||||||
|
|
||||||
|
rep2 := &recReporter{}
|
||||||
|
ok := newEnsemble("x", "x", nil, nil, true, &verdictAdj{})
|
||||||
|
_, err = ok.e.Transcribe(progress.With(context.Background(), rep2), "/a.wav")
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.NotContains(t, rep2.joined(), "fallback", "a healthy run never touches the cloud step")
|
||||||
|
}
|
||||||
|
|
||||||
|
// cancellingBackend ends the job in the middle of its own call, like a shutdown.
|
||||||
|
type cancellingBackend struct {
|
||||||
|
name string
|
||||||
|
cancel context.CancelFunc
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c cancellingBackend) Name() string { return c.name }
|
||||||
|
func (c cancellingBackend) Transcribe(ctx context.Context, _ string) (domain.Transcription, error) {
|
||||||
|
c.cancel()
|
||||||
|
return domain.Transcription{}, ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsembleProgress_AShutdownMidEngineIsNotReportedAsAFailedEngine(t *testing.T) {
|
||||||
|
rep := &recReporter{}
|
||||||
|
ctx, cancel := context.WithCancel(progress.With(context.Background(), rep))
|
||||||
|
calls := &[]string{}
|
||||||
|
alt := &orderedBackend{name: "iguana/kb-whisper", text: "x", calls: calls}
|
||||||
|
e := transcribe.NewEnsemble(cancellingBackend{name: "iguana/pianissimo", cancel: cancel}, alt, nil, &verdictAdj{}, ensembleOpts)
|
||||||
|
|
||||||
|
_, err := e.Transcribe(ctx, "/a.wav")
|
||||||
|
|
||||||
|
require.ErrorIs(t, err, context.Canceled)
|
||||||
|
assert.NotContains(t, rep.joined(), "fail:", "a stopped job is not a broken engine: the page must not blame one")
|
||||||
|
assert.Empty(t, *calls, "and the second engine never starts")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsembleProgress_TheReconcileCountsAreNotedForTheOperator(t *testing.T) {
|
||||||
|
rep := &recReporter{}
|
||||||
|
r := newEnsemble("Totalt 1015 kronor idag.", "Totalt 1050 kronor idag.", nil, nil, false, &verdictAdj{v: reconcile.VerdictAlt})
|
||||||
|
|
||||||
|
_, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, rep.notes, 1)
|
||||||
|
n := rep.notes[0]
|
||||||
|
assert.Equal(t, "reconciled", n["msg"])
|
||||||
|
assert.Equal(t, 1, n["disputes"])
|
||||||
|
assert.Equal(t, 1, n["changed"])
|
||||||
|
assert.Equal(t, 0, n["unsure"])
|
||||||
|
assert.Equal(t, 0, n["errors"])
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user