talk/test/queue_test.go

package test

import (
	"fmt"
	"net/http"
	"strings"
	"testing"
	"time"
)

// TestSerialization: two announcements enqueued in quick succession play one
// after another — the second synthesis reaches the fake provider only AFTER
// the response to the first (no overlap), and the text order stays
// enqueue-stable. The null sink visibly discards every announcement.
func TestSerialization(t *testing.T) {
	root := t.TempDir()
	fake := newFakeProvider(t, 250*time.Millisecond)
	d := startDaemon(t, root, fake.srv.URL, nil)

	code, first := speak(t, d, `{"text":"first announcement"}`)
	if code != 202 || first["id"] == "" || first["muted"] == true {
		t.Fatalf("first: status=%d body=%v, want 202 with id", code, first)
	}

	// As soon as the first synthesis has arrived at the fake (daemon busy),
	// two more in quick succession — they must line up behind the running
	// one.
	fake.waitArrived(t, 1)
	speak(t, d, `{"text":"second announcement"}`)
	speak(t, d, `{"text":"third announcement"}`)

	hits := fake.waitHits(t, 3)

	var texts []string
	for _, h := range hits {
		texts = append(texts, h.Text)
	}
	if strings.Join(texts, "|") != "first announcement|second announcement|third announcement" {
		t.Errorf("playback order = %v, want FIFO", texts)
	}
	// Serialization: the next synthesis starts only after the response to
	// the previous one (small tolerance for clock jitter).
	const slack = 25 * time.Millisecond
	for i := 1; i < len(hits); i++ {
		if hits[i].Start.Before(hits[i-1].End.Add(-slack)) {
			t.Errorf("announcement %d overlaps announcement %d: start %v < end %v", i, i-1, hits[i].Start, hits[i-1].End)
		}
	}

	// Wait for idle first: the third discard line appears daemon-side only
	// shortly AFTER the response recorded at the fake.
	waitStatusIdle(t, d, 3*time.Second)

	// The null sink discarded all three (silent, but visible).
	if n := strings.Count(d.stderr.String(), "discarded (null sink)"); n < 3 {
		t.Errorf("null-sink discards in the daemon log = %d, want >= 3 (stderr: %s)", n, d.stderr.String())
	}
}

// TestPriority: during a long announcement low/normal/high are enqueued —
// processing follows high > normal > low (FIFO per level).
func TestPriority(t *testing.T) {
	root := t.TempDir()
	fake := newFakeProvider(t, 200*time.Millisecond)
	d := startDaemon(t, root, fake.srv.URL, nil)

	speak(t, d, `{"text":"blocker"}`)
	fake.waitArrived(t, 1) // blocker running, worker bound.

	// Enqueued in mixed class order; the queue view reports IDs — so
	// collect text→id and check the order via IDs.
	enq := make(map[string]string)
	for _, tc := range []struct{ body, text string }{
		{`{"text":"low one","priority":"low"}`, "low one"},
		{`{"text":"normal-one"}`, "normal-one"},
		{`{"text":"normal-two"}`, "normal-two"},
		{`{"text":"urgent","priority":"high"}`, "urgent"},
	} {
		code, m := speak(t, d, tc.body)
		if code != 202 {
			t.Fatalf("speak %s: status %d", tc.body, code)
		}
		id, _ := m["id"].(string)
		if id == "" {
			t.Fatalf("speak %s: no ID in %v", tc.body, m)
		}
		enq[tc.text] = id
	}
	wantOrder := []string{"urgent", "normal-one", "normal-two", "low one"}

	// Status shows the current item and the waiting ones in processing
	// order (deterministic: the blocker holds the worker).
	st := getStatus(t, d)
	if st.Current == nil || st.Current.Text != "blocker" {
		t.Fatalf("current = %+v, want blocker", st.Current)
	}
	if st.Queue != 4 {
		t.Fatalf("queue length = %d, want 4", st.Queue)
	}
	var wantIDs []string
	for _, text := range wantOrder {
		wantIDs = append(wantIDs, enq[text])
	}
	if fmt.Sprint(st.QueueIDs) != fmt.Sprint(wantIDs) {
		t.Errorf("queue_ids = %v, want %v (processing order)", st.QueueIDs, wantIDs)
	}

	hits := fake.waitHits(t, 5)
	var texts []string
	for _, h := range hits {
		texts = append(texts, h.Text)
	}
	want := append([]string{"blocker"}, wantOrder...)
	if strings.Join(texts, "|") != strings.Join(want, "|") {
		t.Errorf("playback order = %v, want %v", texts, want)
	}
	waitStatusIdle(t, d, 3*time.Second)
}

// TestNowJumpsAhead: --now via the CLI sets now:true — the announcement
// jumps ahead of all waiting normal items (but does not interrupt the
// running blocker).
func TestNowJumpsAhead(t *testing.T) {
	root := t.TempDir()
	fake := newFakeProvider(t, 600*time.Millisecond)
	d := startDaemon(t, root, fake.srv.URL, nil)
	env := daemonEnv(root, d.port, fake.srv.URL, nil)

	speak(t, d, `{"text":"blocker"}`)
	fake.waitArrived(t, 1)

	speak(t, d, `{"text":"later"}`)

	res := runTalk(t, env, "say", "--now", "right away")
	if res.code != 0 {
		t.Fatalf("talk say --now: exit %d, stderr: %s", res.code, res.stderr)
	}
	if id := strings.TrimSpace(res.stdout); id == "" || strings.Contains(id, "\n") {
		t.Errorf("talk say --now stdout = %q, want one line with the item ID", res.stdout)
	}

	hits := fake.waitHits(t, 3)
	var texts []string
	for _, h := range hits {
		texts = append(texts, h.Text)
	}
	if strings.Join(texts, "|") != "blocker|right away|later" {
		t.Errorf("playback order = %v, want [blocker right away later] — now must jump ahead of normal", texts)
	}
	waitStatusIdle(t, d, 3*time.Second)
}

// TestCancel: aborts the running announcement (synthesis request is
// cancelled), empties the queue, reports the total count — and the daemon
// remains functional afterwards.
func TestCancel(t *testing.T) {
	root := t.TempDir()
	fake := newFakeProvider(t, 5*time.Second)
	d := startDaemon(t, root, fake.srv.URL, nil)

	code, first := speak(t, d, `{"text":"blocker"}`)
	if code != 202 {
		t.Fatalf("speak: status %d", code)
	}
	fake.waitArrived(t, 1) // blocker is stuck in synthesis (5 s fake delay).

	code, second := speak(t, d, `{"text":"waiting"}`)
	if code != 202 {
		t.Fatalf("speak: status %d", code)
	}

	// Status contract: current item + one waiting.
	st := getStatus(t, d)
	if st.Current == nil || st.Current.ID != first["id"] {
		t.Fatalf("current = %+v, want the blocker (%v)", st.Current, first["id"])
	}
	if st.Queue != 1 || len(st.QueueIDs) != 1 || st.QueueIDs[0] != second["id"] {
		t.Fatalf("queue = %v (%d), want the waiting item", st.QueueIDs, st.Queue)
	}

	code, m := postPath(t, d, "/cancel")
	if code != 200 {
		t.Fatalf("POST /cancel: status %d", code)
	}
	if m["cancelled"] != float64(2) {
		t.Fatalf("cancelled = %v, want 2 (running + waiting)", m["cancelled"])
	}

	// The daemon becomes idle immediately (synthesis cancelled) and accepts
	// new announcements — cancel does not damage the daemon.
	waitStatusIdle(t, d, 2*time.Second)
	fake.setDelay(50 * time.Millisecond)
	speak(t, d, `{"text":"afterwards"}`)
	hits := fake.waitCompleted(t, 1)
	if hits[0].Text != "afterwards" {
		t.Errorf("first completed hit = %q, want afterwards (cancelled ones do not count)", hits[0].Text)
	}
	waitStatusIdle(t, d, 3*time.Second)
}

// TestSpeakValidation: the HTTP contract rejects broken requests with 400
// (and 405 for the wrong method) without damaging the daemon.
func TestSpeakValidation(t *testing.T) {
	root := t.TempDir()
	fake := newFakeProvider(t, 0)
	d := startDaemon(t, root, fake.srv.URL, nil)

	cases := []struct {
		name string
		body string
	}{
		{"not JSON", `not json`},
		{"text missing", `{}`},
		{"text empty", `{"text":"   "}`},
		{"priority garbage", `{"text":"x","priority":"immediately"}`},
		{"provider unknown", `{"text":"x","provider":"doesnotexist"}`},
		{"lang garbage", `{"text":"x","lang":"fr"}`},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			code, m := speak(t, d, tc.body)
			if code != 400 {
				t.Fatalf("status = %d, want 400 (body: %v)", code, m)
			}
			if _, ok := m["error"]; !ok {
				t.Errorf("response without error field: %v", m)
			}
		})
	}

	// Wrong method → 405 (method-bound routes).
	resp, err := http.Post(d.baseURL+"/status", "application/json", nil)
	if err != nil {
		t.Fatal(err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusMethodNotAllowed {
		t.Errorf("POST /status: status = %d, want 405", resp.StatusCode)
	}

	// Afterwards the daemon is still healthy and speakable.
	code, m := speak(t, d, `{"text":"healthy"}`)
	if code != 202 || m["id"] == "" {
		t.Fatalf("after validation errors: speak = %d %v, want 202 with id", code, m)
	}
	fake.waitHits(t, 1)
}

// TestStatusContract: fields of the status response when idle (version,
// mute off, no current, provider resolved).
func TestStatusContract(t *testing.T) {
	root := t.TempDir()
	fake := newFakeProvider(t, 0)
	d := startDaemon(t, root, fake.srv.URL, nil)

	st := getStatus(t, d)
	if st.Version != "dev" {
		t.Errorf("version = %q, want dev (built without ldflags)", st.Version)
	}
	if st.Muted {
		t.Error("muted = true on a fresh start, want false")
	}
	if st.Current != nil {
		t.Errorf("current = %+v, want null when idle", st.Current)
	}
	if st.Queue != 0 {
		t.Errorf("queue = %d, want 0", st.Queue)
	}
	// Active provider: elevenlabs first, because key + fake endpoint are set.
	if st.Provider != "elevenlabs" {
		t.Errorf("provider = %q, want elevenlabs", st.Provider)
	}
}