Files
talk/test/queue_test.go
T
m3ta-chiron 1ae1f1262c test: Prozessgrenzen-Tests — talkd + talk als echte Prozesse gegen Fake-Provider und sink null
Primäre Seam laut Spec: talkd/talk gebuildet und als Prozesse gefahren,
ElevenLabs-Fake per httptest (Ankünfte + Antworten mit Zeiten), TALK_PORT
auf freie Ports (nie 7800), isoliertes Config-Dir. Getestet:
Serialisierung/Priorität/now-Sprung, cancel (Synthese-Abbruch + Zahl),
non-blocking + --wait, mute-Persistenz über Restart + offline geschriebene
Datei, Degraded Mode (Lock-Serialisierung, mute, Lock-Freigabe),
3 parallele talk-Aufrufe, HTTP-Validierung (400/405), Status-Vertrag.
2026-10-18 17:10:36 +02:00

274 lines
8.8 KiB
Go

package test
import (
"fmt"
"net/http"
"strings"
"testing"
"time"
)
// TestSerialisierung: zwei schnell hintereinander eingereihte Ansagen
// spielen nacheinander — die zweite Synthese erreicht den Fake-Provider
// erst NACH der Antwort der ersten (kein Überlappen), und die Textfolge
// bleibt enqueue-stabil. Der null-Sink verwirft jede Ansage sichtbar.
func TestSerialisierung(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":"erste ansage"}`)
if code != 202 || first["id"] == "" || first["muted"] == true {
t.Fatalf("erste: Status=%d Body=%v, want 202 mit id", code, first)
}
// Sobald die erste Synthese beim Fake angekommen ist (Daemon
// beschäftigt), zwei weitere schnell hinterher — sie müssen sich
// hinter der laufenden einreihen.
fake.waitArrived(t, 1)
speak(t, d, `{"text":"zweite ansage"}`)
speak(t, d, `{"text":"dritte ansage"}`)
hits := fake.waitHits(t, 3)
var texts []string
for _, h := range hits {
texts = append(texts, h.Text)
}
if strings.Join(texts, "|") != "erste ansage|zweite ansage|dritte ansage" {
t.Errorf("Abspiel-Reihenfolge = %v, want FIFO", texts)
}
// Serialisierung: die nächste Synthese startet erst nach der Antwort
// der vorherigen (kleine Toleranz für Uhr-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("Ansage %d überlappt mit Ansage %d: Start %v < Ende %v", i, i-1, hits[i].Start, hits[i-1].End)
}
}
// Erst idle warten: Die dritte Verwurf-Zeile entsteht daemon-seitig
// erst kurz NACH der beim Fake aufgezeichneten Antwort.
waitStatusIdle(t, d, 3*time.Second)
// Der null-Sink hat alle drei verworfen (tonlos, aber sichtbar).
if n := strings.Count(d.stderr.String(), "verworfen (null sink)"); n < 3 {
t.Errorf("null-Sink-Verwürfe im Daemon-Log = %d, want >= 3 (stderr: %s)", n, d.stderr.String())
}
}
// TestPrioritaet: während einer langen Ansage werden low/normal/high
// eingereiht — die Abarbeitung folgt high > normal > low (FIFO je Level).
func TestPrioritaet(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 läuft, Worker gebunden.
// Eingereiht wird in Klassen-Durcheinander; die Queue-Sicht meldet
// IDs — also text→id einsammeln und die Reihenfolge über IDs prüfen.
enq := make(map[string]string)
for _, tc := range []struct{ body, text string }{
{`{"text":"niedrig","priority":"low"}`, "niedrig"},
{`{"text":"normal-eins"}`, "normal-eins"},
{`{"text":"normal-zwei"}`, "normal-zwei"},
{`{"text":"dringend","priority":"high"}`, "dringend"},
} {
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: keine ID in %v", tc.body, m)
}
enq[tc.text] = id
}
wantOrder := []string{"dringend", "normal-eins", "normal-zwei", "niedrig"}
// Status zeigt aktuelles Item und die wartenden in Abarbeitungs-
// Reihenfolge (deterministisch: der Blocker hält den Worker).
st := getStatus(t, d)
if st.Current == nil || st.Current.Text != "blocker" {
t.Fatalf("current = %+v, will blocker", st.Current)
}
if st.Queue != 4 {
t.Fatalf("Queue-Länge = %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 (Abarbeitungs-Reihenfolge)", 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("Abspiel-Reihenfolge = %v, want %v", texts, want)
}
waitStatusIdle(t, d, 3*time.Second)
}
// TestNowSpringtVor: --now über die CLI setzt now:true — die Ansage
// springt vor alle wartenden normal-Items (unterbricht aber nicht den
// laufenden Blocker).
func TestNowSpringtVor(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":"spaeter"}`)
res := runTalk(t, env, "say", "--now", "sofort")
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, will eine Zeile mit der 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|sofort|spaeter" {
t.Errorf("Abspiel-Reihenfolge = %v, want [blocker sofort spaeter] — now muss vor normal springen", texts)
}
waitStatusIdle(t, d, 3*time.Second)
}
// TestCancel: bricht die laufende Ansage ab (Synthese-Request wird
// abgebrochen), leert die Queue, meldet die Gesamtzahl — und der Daemon
// bleibt danach funktionstüchtig.
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 steckt in der Synthese (5 s Fake-Delay).
code, second := speak(t, d, `{"text":"wartet"}`)
if code != 202 {
t.Fatalf("speak: Status %d", code)
}
// Status-Vertrag: aktuelles Item + ein wartendes.
st := getStatus(t, d)
if st.Current == nil || st.Current.ID != first["id"] {
t.Fatalf("current = %+v, will den Blocker (%v)", st.Current, first["id"])
}
if st.Queue != 1 || len(st.QueueIDs) != 1 || st.QueueIDs[0] != second["id"] {
t.Fatalf("queue = %v (%d), will das wartende 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 (laufende + wartende)", m["cancelled"])
}
// Der Daemon wird sofort idle (Synthese abgebrochen) und nimmt neue
// Ansagen an — cancel beschädigt den Daemon nicht.
waitStatusIdle(t, d, 2*time.Second)
fake.setDelay(50 * time.Millisecond)
speak(t, d, `{"text":"nachher"}`)
hits := fake.waitCompleted(t, 1)
if hits[0].Text != "nachher" {
t.Errorf("erster abgeschlossener Treffer = %q, will nachher (abgebrochene zählen nicht)", hits[0].Text)
}
waitStatusIdle(t, d, 3*time.Second)
}
// TestSpeakValidierung: der HTTP-Vertrag lehnt kaputte Anfragen mit 400
// ab (und 405 bei falscher Methode), ohne den Daemon zu beschädigen.
func TestSpeakValidierung(t *testing.T) {
root := t.TempDir()
fake := newFakeProvider(t, 0)
d := startDaemon(t, root, fake.srv.URL, nil)
cases := []struct {
name string
body string
}{
{"kein JSON", `kein json`},
{"text fehlt", `{}`},
{"text leer", `{"text":" "}`},
{"priority Müll", `{"text":"x","priority":"sofort"}`},
{"provider unbekannt", `{"text":"x","provider":"gibtsnicht"}`},
{"lang Müll", `{"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("Antwort ohne error-Feld: %v", m)
}
})
}
// Falsche Methode → 405 (Methoden-gebundene Routen).
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)
}
// Danach ist der Daemon weiterhin gesund und speakbar.
code, m := speak(t, d, `{"text":"gesund"}`)
if code != 202 || m["id"] == "" {
t.Fatalf("nach Validierungs-Fehlern: speak = %d %v, want 202 mit id", code, m)
}
fake.waitHits(t, 1)
}
// TestStatusVertrag: Felder der Status-Antwort im Leerlauf (Version,
// Mute aus, kein current, Provider aufgelöst).
func TestStatusVertrag(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 (ohne ldflags gebaut)", st.Version)
}
if st.Muted {
t.Error("muted = true im Frischstart, want false")
}
if st.Current != nil {
t.Errorf("current = %+v, want null im Leerlauf", st.Current)
}
if st.Queue != 0 {
t.Errorf("queue = %d, want 0", st.Queue)
}
// Aktiver Provider: elfenlabs zuerst, weil Key + Fake-Endpunkt gesetzt.
if st.Provider != "elevenlabs" {
t.Errorf("provider = %q, want elevenlabs", st.Provider)
}
}