200 lines
6.7 KiB
Go
200 lines
6.7 KiB
Go
package uploads_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
|
"git.d-ma.be/mathias/tapir/internal/progress"
|
|
"git.d-ma.be/mathias/tapir/internal/uploads"
|
|
)
|
|
|
|
// Progress, heartbeat and logging of the worker (ADR-035, #42, #43).
|
|
|
|
// plannedTr is a transcriber that declares its own phases and reports them, like the
|
|
// two-engine Ensemble does.
|
|
type plannedTr struct{ *fakeTranscriber }
|
|
|
|
func (plannedTr) Plan() []progress.Planned {
|
|
return []progress.Planned{{Key: "engine_a", Label: "Transcribing with A"}, {Key: "engine_b", Label: "Transcribing with B"}}
|
|
}
|
|
|
|
func (p plannedTr) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) {
|
|
r := progress.From(ctx)
|
|
r.Begin("engine_a", 0)
|
|
r.Done("engine_a")
|
|
r.Begin("engine_b", 0)
|
|
r.Done("engine_b")
|
|
return p.fakeTranscriber.Transcribe(ctx, wav)
|
|
}
|
|
|
|
func keysAndStates(steps []domain.ProgressStep) []string {
|
|
var out []string
|
|
for _, s := range steps {
|
|
out = append(out, s.Key+":"+string(s.State))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// syncLog is a log sink safe to read while the worker's goroutines write.
|
|
type syncLog struct {
|
|
mu sync.Mutex
|
|
b bytes.Buffer
|
|
}
|
|
|
|
func (l *syncLog) Write(p []byte) (int, error) {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
return l.b.Write(p)
|
|
}
|
|
|
|
func (l *syncLog) String() string {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
return l.b.String()
|
|
}
|
|
|
|
func newProgressWorker(t *testing.T, r *wrig, tr uploads.Transcriber, opts ...uploads.WorkerOption) *uploads.Worker {
|
|
t.Helper()
|
|
return uploads.NewWorker(r.st, r.conv, tr, r.tmp, time.Minute, opts...)
|
|
}
|
|
|
|
func TestWorkerProgress_ASingleEngineJobShowsEveryPhaseDone(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
r.w.Process(context.Background(), r.spool(t, "a"))
|
|
|
|
assert.Equal(t, []string{"received:done", "audio:done", "transcribe:done", "save:done"}, keysAndStates(r.st.lastSteps()))
|
|
}
|
|
|
|
func TestWorkerProgress_ATranscriberThatPlansItsOwnPhasesGetsThemBetweenAudioAndSaving(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
w := newProgressWorker(t, r, plannedTr{r.tr})
|
|
|
|
w.Process(context.Background(), r.spool(t, "a"))
|
|
|
|
assert.Equal(t, []string{"received:done", "audio:done", "engine_a:done", "engine_b:done", "save:done"}, keysAndStates(r.st.lastSteps()))
|
|
}
|
|
|
|
func TestWorkerProgress_AFailureMarksWhereItStoppedAndLeavesTheRestPending(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
setup func(*wrig)
|
|
want []string
|
|
}{
|
|
{
|
|
"conversion fails", func(r *wrig) { r.conv.err = errors.New("moov atom not found") },
|
|
[]string{"received:done", "audio:failed", "transcribe:pending", "save:pending"},
|
|
},
|
|
{
|
|
"every engine fails", func(r *wrig) { r.tr.err = errors.New("iguana down") },
|
|
[]string{"received:done", "audio:done", "transcribe:failed", "save:pending"},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
tt.setup(r)
|
|
|
|
r.w.Process(context.Background(), r.spool(t, "a"))
|
|
|
|
assert.Equal(t, tt.want, keysAndStates(r.st.lastSteps()))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestWorkerProgress_AShutdownIsNotRecordedAsAFailedStep(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
r.tr.block = make(chan struct{})
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { defer close(done); r.w.Process(ctx, r.spool(t, "a")) }()
|
|
require.Eventually(t, func() bool {
|
|
return strings.Contains(strings.Join(keysAndStates(r.st.lastSteps()), ","), "transcribe:running")
|
|
}, 2*time.Second, time.Millisecond)
|
|
|
|
cancel()
|
|
<-done
|
|
|
|
assert.Contains(t, keysAndStates(r.st.lastSteps()), "transcribe:running",
|
|
"a shutdown leaves the row for the start-up sweep; it must not pose as a failure of the step")
|
|
}
|
|
|
|
func TestWorkerProgress_TheHeartbeatBeatsWhileAnEngineCallIsStuckAndStopsWhenTheJobEnds(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
r.tr.block = make(chan struct{})
|
|
r.st.beatDelay = 20 * time.Millisecond
|
|
w := newProgressWorker(t, r, r.tr, uploads.WithHeartbeat(5*time.Millisecond, time.Hour))
|
|
done := make(chan struct{})
|
|
go func() { defer close(done); w.Process(context.Background(), r.spool(t, "a")) }()
|
|
|
|
require.Eventually(t, func() bool { return r.st.beatCount() >= 3 }, 2*time.Second, 5*time.Millisecond,
|
|
"the page tells a slow engine from a dead service by this heartbeat, so it must not wait on the engine")
|
|
close(r.tr.block)
|
|
<-done
|
|
after := r.st.beatCount()
|
|
time.Sleep(40 * time.Millisecond)
|
|
|
|
assert.Equal(t, after, r.st.beatCount(), "no heartbeat may outlive its job")
|
|
}
|
|
|
|
func TestWorkerProgress_OnASuccessfulJobTheLogNamesEveryPhaseInOrderAndNeverTheTranscriptOrThePath(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
r.tr.result = domain.Transcription{Text: "zebrakaka secret meeting words", Endpoint: "iguana/kb-whisper", AudioSeconds: 3}
|
|
buf := &syncLog{}
|
|
w := newProgressWorker(t, r, r.tr, uploads.WithLogger(slog.New(slog.NewTextHandler(buf, nil))))
|
|
|
|
w.Process(context.Background(), r.spool(t, "a"))
|
|
|
|
out := buf.String()
|
|
audio := strings.Index(out, "step=audio")
|
|
transcribe := strings.Index(out, "step=transcribe")
|
|
save := strings.Index(out, "step=save")
|
|
require.True(t, audio >= 0 && transcribe > audio && save > transcribe, "phases must be logged in the order they ran:\n%s", out)
|
|
assert.Contains(t, out, "upload=a", "every line says which upload it is about")
|
|
assert.NotContains(t, out, "zebrakaka", "no transcript text in the log")
|
|
assert.NotContains(t, out, r.tmp, "no spool path, which carries the upload's identity, in the log")
|
|
assert.Contains(t, out, "upload transcribed", "the job's own summary line")
|
|
assert.Contains(t, out, "endpoint=iguana/kb-whisper", "which endpoint served it is a quality fact")
|
|
}
|
|
|
|
func TestWorkerProgress_AFailureLogsAKindNeverTheErrorTextOrThePath(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
setup func(*wrig)
|
|
line string
|
|
}{
|
|
{"conversion fails and names the spool file", func(r *wrig) {
|
|
r.conv.err = errors.New("Error opening input file " + r.tmp + "/upload-a: Invalid data zebrakaka")
|
|
}, "upload: convert"},
|
|
{"every engine fails and the gateway echoed a word", func(r *wrig) {
|
|
r.tr.err = errors.New("gateway echoed: zebrakaka secret words")
|
|
}, "upload: transcribe"},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
r := newWRig(t, time.Minute)
|
|
tt.setup(r)
|
|
buf := &syncLog{}
|
|
w := newProgressWorker(t, r, r.tr, uploads.WithLogger(slog.New(slog.NewTextHandler(buf, nil))))
|
|
|
|
w.Process(context.Background(), r.spool(t, "a"))
|
|
|
|
out := buf.String()
|
|
assert.NotContains(t, out, "zebrakaka", "provider text or file contents in the log")
|
|
assert.NotContains(t, out, r.tmp, "the spool path in the log")
|
|
assert.Contains(t, out, tt.line, "the failure must be logged at all, not only scrubbed")
|
|
assert.Contains(t, out, "kind=error")
|
|
assert.Contains(t, out, "upload=a")
|
|
})
|
|
}
|
|
}
|