tapir/cmd/tapir/e2e_upload_test.go

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")
	if os.Getenv("TAPIR_E2E_RECONCILE") == "1" {
		// Two engines and a local model, as ADR-034 describes.
		cfg.ReconcileBaseModel = envOrDefault("TAPIR_E2E_BASE_MODEL", "iguana/pianissimo")
		cfg.ReconcileModel = envOrDefault("TAPIR_E2E_RECONCILE_MODEL", "iguana/gemma4-26b")
	}
	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"}},
		Chat: buildChat(cfg),
	}
	wireUploads(app, cfg, intake) // the same wiring cmdServe uses
	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 func() { _ = 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
	var hadVideo bool
	require.NoError(t, raw.QueryRow(ctx,
		`SELECT status, endpoint, content, audio_seconds, had_video FROM uploaded_transcripts WHERE user_id = $1`, uid).
		Scan(&status, &endpoint, &content, &secs, &hadVideo))
	t.Logf("done: endpoint=%s audio_s=%.1f transcript_chars=%d had_video=%v", endpoint, secs, len(content), hadVideo)
	if os.Getenv("TAPIR_E2E_RECONCILE") == "1" {
		var recJSON string
		require.NoError(t, raw.QueryRow(ctx,
			`SELECT COALESCE(reconciliation::text, 'NULL') FROM uploaded_transcripts WHERE user_id = $1`, uid).Scan(&recJSON))
		require.NotEqual(t, "NULL", recJSON, "a two-engine run must record its reconciliation")
		t.Logf("reconciliation: %.600s", recJSON)
		assert.Contains(t, endpoint, "reconciled by", "every contributor is named in the endpoint")
		assert.Contains(t, page, "Reconciled from two transcripts")
	}
	if os.Getenv("TAPIR_E2E_HAD_VIDEO") == "1" {
		// A Teams meeting with the camera off: a video file whose audio is what matters.
		assert.True(t, hadVideo, "the file had a video track and the row must say so")
		assert.Contains(t, page, "Only its audio was used", "and the user must be told")
	}
	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)"
}