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