diff --git a/cmd/tapir/e2e_upload_test.go b/cmd/tapir/e2e_upload_test.go new file mode 100644 index 0000000..0cf9ecc --- /dev/null +++ b/cmd/tapir/e2e_upload_test.go @@ -0,0 +1,170 @@ +package main + +import ( + "context" + "fmt" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "git.d-ma.be/mathias/tapir/internal/adapters/store" + "git.d-ma.be/mathias/tapir/internal/config" + "git.d-ma.be/mathias/tapir/internal/web" +) + +// TestE2E_UploadTranscribeAndChat is the end-to-end check for tapir#33: upload a +// real audio file, watch it become a transcript through the REAL ffprobe, ffmpeg +// and gateway, then ask a question and get a grounded answer from a REAL model. +// It is opt-in — it needs real binaries and the live gateway — and skipped unless +// TAPIR_E2E_AUDIO is set: +// +// TAPIR_E2E_AUDIO=/path/meeting.m4a TAPIR_E2E_FFMPEG=/path/ffmpeg \ +// TAPIR_E2E_FFPROBE=/path/ffprobe TAPIR_E2E_QUESTION='Vem ansvarar för virket?' \ +// TAPIR_E2E_EXPECT=anders op run --env-file ~/.op-env -- \ +// go test ./cmd/tapir -run E2E_Upload -v -count=1 +// +// Use SYNTHETIC audio only: this repo is public, and a fallback to berget sends +// the audio to a cloud endpoint (#33). +func TestE2E_UploadTranscribeAndChat(t *testing.T) { + audioPath := os.Getenv("TAPIR_E2E_AUDIO") + if audioPath == "" { + t.Skip("set TAPIR_E2E_AUDIO (and friends) to run the live end-to-end check") + } + key := os.Getenv("LITELLM_MASTER_KEY") + require.NotEmpty(t, key, "LITELLM_MASTER_KEY must be set (via op run)") + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + // A private embedded Postgres: nothing here may touch a real database. + port := uint32(55000 + os.Getpid()%1000) + rt := filepath.Join(os.TempDir(), fmt.Sprintf("tapir-epg-e2e-%d", os.Getpid())) + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig().Port(port). + RuntimePath(rt).DataPath(filepath.Join(rt, "data")).BinariesPath(filepath.Join(rt, "bin")). + CachePath(filepath.Join(os.TempDir(), "tapir-epg-cache"))) + require.NoError(t, pg.Start()) + t.Cleanup(func() { _ = pg.Stop(); _ = os.RemoveAll(rt) }) + dsn := fmt.Sprintf("postgres://postgres:postgres@localhost:%d/postgres?sslmode=disable", port) + + st, err := store.New(ctx, dsn) + require.NoError(t, err) + t.Cleanup(st.Close) + raw, err := pgxpool.New(ctx, dsn) + require.NoError(t, err) + t.Cleanup(raw.Close) + const uid = "11111111-1111-1111-1111-111111111111" + _, err = raw.Exec(ctx, `INSERT INTO users (id) VALUES ($1)`, uid) + require.NoError(t, err) + _, err = raw.Exec(ctx, `INSERT INTO user_identities (dex_subject, user_id) VALUES ('e2e', $1)`, uid) + require.NoError(t, err) + + cfg, err := config.Load() + require.NoError(t, err) + cfg.GatewayKey = key + cfg.UploadsEnabled = true + cfg.UploadTmpDir = filepath.Join(t.TempDir(), "spool") + cfg.FFmpegBin = envOrDefault("TAPIR_E2E_FFMPEG", "ffmpeg") + cfg.FFprobeBin = envOrDefault("TAPIR_E2E_FFPROBE", "ffprobe") + + intake, worker, err := buildUploads(ctx, cfg, st, quietLog()) + require.NoError(t, err) + go worker.Run(ctx) + app := &web.App{ + Store: st, Identity: st, Auth: web.StubAuth{U: web.User{Subject: "e2e"}}, + Uploads: intake, UploadLimits: uploadLimits(cfg), + Chat: buildChat(cfg), UploadChatMaxChars: cfg.MaxUploadedTranscriptChars, + } + srv := httptest.NewServer(app.Router()) + t.Cleanup(srv.Close) + noRedirect := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + + // 1. Upload. + f, err := os.Open(audioPath) + require.NoError(t, err) + defer f.Close() + pr, pw := io.Pipe() + mw := multipart.NewWriter(pw) + go func() { + _ = mw.WriteField("title", "E2E recording") + fw, _ := mw.CreateFormFile("file", filepath.Base(audioPath)) + _, _ = io.Copy(fw, f) + _ = mw.Close() + _ = pw.Close() + }() + req, err := http.NewRequest(http.MethodPost, srv.URL+"/uploads", pr) + require.NoError(t, err) + req.Header.Set("Content-Type", mw.FormDataContentType()) + resp, err := noRedirect.Do(req) + require.NoError(t, err) + _ = resp.Body.Close() + require.Equal(t, http.StatusSeeOther, resp.StatusCode, "an accepted upload redirects to its page") + loc := resp.Header.Get("Location") + require.True(t, strings.HasPrefix(loc, "/uploads/"), loc) + t.Logf("accepted -> %s", loc) + + // 2. Wait for the transcript: unfinished states poll, terminal states do not. + var page string + require.Eventually(t, func() bool { + r, err := http.Get(srv.URL + loc + "/status") + if err != nil { + return false + } + b, _ := io.ReadAll(r.Body) + _ = r.Body.Close() + page = string(b) + return !strings.Contains(page, "hx-trigger") + }, 12*time.Minute, 3*time.Second, "the upload never reached a terminal state") + require.Contains(t, page, "Transcribed", "terminal state was not 'done': %.300s", page) + + var status, endpoint, content string + var secs float64 + require.NoError(t, raw.QueryRow(ctx, + `SELECT status, endpoint, content, audio_seconds FROM uploaded_transcripts WHERE user_id = $1`, uid). + Scan(&status, &endpoint, &content, &secs)) + t.Logf("done: endpoint=%s audio_s=%.1f transcript_chars=%d", endpoint, secs, len(content)) + assert.Equal(t, "done", status) + assert.NotEmpty(t, endpoint) + assert.NotEmpty(t, content) + entries, err := os.ReadDir(cfg.UploadTmpDir) + require.NoError(t, err) + assert.Empty(t, entries, "the audio must be deleted once transcribed") + + // 3. Ask a question; the answer must come from the transcript. + q := envOrDefault("TAPIR_E2E_QUESTION", "Who is responsible for the timber order?") + chatResp, err := http.PostForm(srv.URL+loc+"/chat", url.Values{"question": {q}}) + require.NoError(t, err) + answerPage, _ := io.ReadAll(chatResp.Body) + _ = chatResp.Body.Close() + require.Equal(t, http.StatusOK, chatResp.StatusCode) + t.Logf("question: %s", q) + if want := os.Getenv("TAPIR_E2E_EXPECT"); want != "" { + assert.Contains(t, strings.ToLower(string(answerPage)), strings.ToLower(want), + "the answer should be grounded in the transcript") + } + t.Logf("answer page excerpt: %.600s", afterMarker(string(answerPage), `class="chat-turn chat-a"`)) +} + +func envOrDefault(k, d string) string { + if v := os.Getenv(k); v != "" { + return v + } + return d +} + +func afterMarker(s, m string) string { + if i := strings.Index(s, m); i >= 0 { + return s[i:] + } + return "(no answer turn found)" +}