// Package test enthält die Prozessgrenzen-Tests von talk 2.0 — die primäre // Test-Seam laut Spec: talkd und talk laufen als echte Prozesse (gebuildet // aus diesem Repo), angetrieben über HTTP und die CLI, mit zwei injizierten // Fakes: einem ElevenLabs-Fake-Endpunkt (httptest-Server mit Audio-Bytes) // und audio.sink=null. Getestet wird nur extern beobachtbares Verhalten: // HTTP-Antworten, CLI-Exit-Codes/-Output, Fake-Treffer (Reihenfolge und // Zeit), Dateien im Config-Dir. Tests binden NIEMALS den Default-Port 7800 // und hängen nie von einem echten Daemon ab — talkd bekommt einen freien // Port per TALK_PORT, CLI-Prozesse denselben (oder Port 1 für „Daemon tot"). package test import ( "bytes" "encoding/json" "fmt" "net" "net/http" "net/http/httptest" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "testing" "time" ) // Die Binaries werden einmal pro Testlauf gebaut (Build-Cache macht es // billig) und von allen Tests geteilt. var ( talkBin string talkdBin string ) func TestMain(m *testing.M) { dir, err := os.MkdirTemp("", "talk-bins-*") if err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } repo, err := filepath.Abs("..") if err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } for _, b := range []struct{ out, pkg string }{ {"talk", "./cmd/talk"}, {"talkd", "./cmd/talkd"}, } { cmd := exec.Command("go", "build", "-o", filepath.Join(dir, b.out), b.pkg) cmd.Dir = repo if out, err := cmd.CombinedOutput(); err != nil { fmt.Fprintf(os.Stderr, "go build %s: %v\n%s", b.pkg, err, out) os.RemoveAll(dir) os.Exit(1) } } talkBin, talkdBin = filepath.Join(dir, "talk"), filepath.Join(dir, "talkd") code := m.Run() os.RemoveAll(dir) os.Exit(code) } // syncBuffer ist ein nebenläufigkeitssicherer Buffer (Daemon-stderr wird // gelesen, während der Prozess läuft). type syncBuffer struct { mu sync.Mutex buf bytes.Buffer } func (b *syncBuffer) Write(p []byte) (int, error) { b.mu.Lock() defer b.mu.Unlock() return b.buf.Write(p) } func (b *syncBuffer) String() string { b.mu.Lock() defer b.mu.Unlock() return b.buf.String() } // fakeHit ist ein beobachteter Synthese-Request am Fake-Provider. type fakeHit struct { Text string // Text aus dem Request-Body Start time.Time // Ankunft am Fake End time.Time // Antwort abgeschickt bzw. Abbruch gesehen Completed bool // true: Antwort geschrieben; false: Client hat abgebrochen } // fakeProvider ist ein ElevenLabs-Fake (POST /v1/text-to-speech/): // Er antwortet mit ein paar Bytes „Audio" und protokolliert jeden Request // mit Ankunfts- und Antwort-Zeit — daraus sind Reihenfolge (Textfolge) und // Serialisierung (Ankunft des nächsten nach Antwort des vorigen) ablesbar. // delay ist zur Laufzeit umschaltbar, um blockierende Synthesen zu // simulieren (Serialisierung, Priorität, cancel). Ankünfte werden separat // und sofort protokolliert: Sie zeigen, dass der Daemon mit einer Synthese // BESCHÄFTIGT ist (Queue deterministisch befüllbar), noch bevor die // (verzögerte) Antwort als Treffer landet. type fakeProvider struct { srv *httptest.Server mu sync.Mutex hits []fakeHit arrived []string delay time.Duration } func newFakeProvider(t *testing.T, delay time.Duration) *fakeProvider { t.Helper() f := &fakeProvider{delay: delay} f.srv = httptest.NewServer(http.HandlerFunc(f.serve)) t.Cleanup(f.srv.Close) return f } func (f *fakeProvider) serve(w http.ResponseWriter, r *http.Request) { var body struct { Text string `json:"text"` } start := time.Now() if err := json.NewDecoder(r.Body).Decode(&body); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } f.mu.Lock() f.arrived = append(f.arrived, body.Text) d := f.delay f.mu.Unlock() if d > 0 { select { case <-time.After(d): case <-r.Context().Done(): // Client (Daemon) hat abgebrochen f.record(fakeHit{Text: body.Text, Start: start, End: time.Now()}) return } } _, _ = w.Write([]byte("\xff\xfb\x90\x00 fake-mp3-bytes")) f.record(fakeHit{Text: body.Text, Start: start, End: time.Now(), Completed: true}) } func (f *fakeProvider) record(h fakeHit) { f.mu.Lock() defer f.mu.Unlock() f.hits = append(f.hits, h) } func (f *fakeProvider) setDelay(d time.Duration) { f.mu.Lock() f.delay = d f.mu.Unlock() } func (f *fakeProvider) snapshot() []fakeHit { f.mu.Lock() defer f.mu.Unlock() out := make([]fakeHit, len(f.hits)) copy(out, f.hits) return out } // completed liefert nur die abgeschlossenen (nicht abgebrochenen) Treffer. func completed(hits []fakeHit) []fakeHit { var out []fakeHit for _, h := range hits { if h.Completed { out = append(out, h) } } return out } // waitArrived pollt, bis n Synthese-Requests angekommen sind — der Daemon // ist damit nachweisbar beschäftigt und die Queue deterministisch befüllbar. func (f *fakeProvider) waitArrived(t *testing.T, n int) { t.Helper() deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { f.mu.Lock() count := len(f.arrived) f.mu.Unlock() if count >= n { return } time.Sleep(10 * time.Millisecond) } t.Fatalf("nur %d von %d Ankünften am Fake innerhalb 5 s", len(f.snapshot()), n) } // arrivedCount zählt die angekommten Synthese-Requests (auch abgebrochene) // — Beobachtungspunkt für „genau N spielen, der Rest parkt/verfällt". func (f *fakeProvider) arrivedCount() int { f.mu.Lock() defer f.mu.Unlock() return len(f.arrived) } // assertNoFurtherArrivals prüft, dass innerhalb der Frist kein weiterer // Synthese-Request ankommt (geparkte/verworfene Ansagen bleiben stumm). func (f *fakeProvider) assertNoFurtherArrivals(t *testing.T, wait time.Duration) { t.Helper() n := f.arrivedCount() time.Sleep(wait) if got := f.arrivedCount(); got != n { t.Fatalf("%d zusätzliche Ankünfte (want keine) — Treffer: %+v", got-n, f.snapshot()) } } // waitHits pollt, bis n Requests abgeschlossen wurden (Abbrüche zählen). func (f *fakeProvider) waitHits(t *testing.T, n int) []fakeHit { t.Helper() deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { if hits := f.snapshot(); len(hits) >= n { return hits } time.Sleep(20 * time.Millisecond) } t.Fatalf("nur %d von %d Fake-Treffern innerhalb 5 s (Treffer: %+v)", len(f.snapshot()), n, f.snapshot()) return nil } // waitCompleted pollt, bis n Requests ERFOLGREICH abgeschlossen wurden // (abgebrochene Synthesen zählen nicht). func (f *fakeProvider) waitCompleted(t *testing.T, n int) []fakeHit { t.Helper() deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { if hits := completed(f.snapshot()); len(hits) >= n { return hits } time.Sleep(20 * time.Millisecond) } t.Fatalf("nur %d von %d abgeschlossenen Treffern innerhalb 5 s (alle: %+v)", len(completed(f.snapshot())), n, f.snapshot()) return nil } // assertNoHits prüft, dass innerhalb der Frist kein Request ankam. func (f *fakeProvider) assertNoHits(t *testing.T, wait time.Duration) { t.Helper() time.Sleep(wait) if hits := f.snapshot(); len(hits) > 0 { t.Fatalf("unerwartete Fake-Treffer: %+v", hits) } } // daemonProc ist ein laufender talkd-Prozess. type daemonProc struct { cmd *exec.Cmd port int stderr *syncBuffer baseURL string } // startDaemon startet talkd als echten Prozess (freier Port per TALK_PORT, // nie der Default-Port) und wartet, bis GET /status antwortet. Stop per // t.Cleanup: SIGINT, nach Frist Kill. func startDaemon(t *testing.T, root, fakeURL string, extra map[string]string) *daemonProc { t.Helper() port := freePort(t) env := daemonEnv(root, port, fakeURL, extra) stderr := &syncBuffer{} cmd := exec.Command(talkdBin) cmd.Env = env cmd.Stderr = stderr if err := cmd.Start(); err != nil { t.Fatalf("talkd starten: %v", err) } d := &daemonProc{ cmd: cmd, port: port, stderr: stderr, baseURL: fmt.Sprintf("http://127.0.0.1:%d", port), } client := &http.Client{Timeout: time.Second} deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { resp, err := client.Get(d.baseURL + "/status") if err == nil { resp.Body.Close() if resp.StatusCode == http.StatusOK { t.Cleanup(func() { d.stop(t) }) return d } } time.Sleep(25 * time.Millisecond) } t.Fatalf("talkd antwortet nicht auf %s — stderr:\n%s", d.baseURL+"/status", stderr.String()) return nil } // stop fährt den Daemon-Prozess herunter (SIGINT wie ein menschlicher // Stop; nach 3 s Kill-Fallback) und wartet auf sein Ende. func (d *daemonProc) stop(t *testing.T) { t.Helper() if d.cmd.Process == nil { return } done := make(chan error, 1) go func() { done <- d.cmd.Wait() }() _ = d.cmd.Process.Signal(os.Interrupt) select { case <-done: case <-time.After(3 * time.Second): _ = d.cmd.Process.Kill() <-done } } // daemonEnv baut die Prozess-Umgebung für talk- und talkd-Prozesse: // geerbtes Environment ohne fremde TALK_*-Variablen (Isolation von Keys/ // Ports der Außenwelt), isoliertes HOME/XDG_CONFIG_HOME (Config-Dir liegt // damit unter /talk), audio.sink=null und der Fake-Provider als // ElevenLabs-Endpunkt. port=1 heißt „garantiert toter Daemon" (Verbindung // wird sofort abgelehnt). // // TALK_RATE_LIMIT=0s ist der Basis-Default der Prozessgrenzen-Tests: Die // Task-04-Semantik-Tests (Serialisierung, Priorität, …) enqueueen dicht // hintereinander und erwarten sofortiges Durchspielen — das Gesetz // (Default 30 s) würde sie parken. Die Gesetz-Tests in ratelimit_test.go // überschreiben den Wert bewusst per extra. func daemonEnv(root string, port int, fakeURL string, extra map[string]string) []string { env := make([]string, 0, 32) for _, kv := range os.Environ() { if strings.HasPrefix(kv, "TALK_") { continue } env = append(env, kv) } set := map[string]string{ "HOME": root, "XDG_CONFIG_HOME": root, "TALK_AUDIO_SINK": "null", "TALK_ELEVENLABS_KEY": "prozesstest-key", "TALK_ELEVENLABS_ENDPOINT": fakeURL, "TALK_EDGE_ENDPOINT": "ws://127.0.0.1:1/", // Kette soll nie auf echte Fremd-Endpunkte fallen "TALK_PORT": strconv.Itoa(port), "TALK_RATE_LIMIT": "0s", // Gesetz aus, bis ein Test es anstellt (siehe oben) } for k, v := range extra { set[k] = v } for k, v := range set { env = append(env, k+"="+v) } return env } // freePort besorgt einen freien Loopback-Port (bind-and-close; der Daemon // bindet ihn sofort danach neu). func freePort(t *testing.T) int { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("freien Port finden: %v", err) } defer ln.Close() return ln.Addr().(*net.TCPAddr).Port } // configDir ist das Config-Verzeichnis unter dem isolierten Root // (Linux-Pfadlogik: XDG_CONFIG_HOME/ → /talk). func configDir(root string) string { return filepath.Join(root, "talk") } // talkResult ist das Beobachtungsergebnis eines talk-Prozesses. type talkResult struct { code int stdout string stderr string } // runTalk führt die talk-CLI als echten Prozess aus. func runTalk(t *testing.T, env []string, args ...string) talkResult { t.Helper() cmd := exec.Command(talkBin, args...) cmd.Env = env var out, errBuf syncBuffer cmd.Stdout = &out cmd.Stderr = &errBuf err := cmd.Run() code := 0 if ee, ok := err.(*exec.ExitError); ok { code = ee.ExitCode() } else if err != nil { t.Fatalf("talk ausführen: %v (stderr: %s)", err, errBuf.String()) } return talkResult{code: code, stdout: out.String(), stderr: errBuf.String()} } // statusBody ist die dekodierte GET /status-Antwort. type statusBody struct { Version string `json:"version"` Queue int `json:"queue"` QueueIDs []string `json:"queue_ids"` Current *struct { ID string `json:"id"` Text string `json:"text"` Priority string `json:"priority"` } `json:"current"` Muted bool `json:"muted"` Provider string `json:"provider"` } // getStatus ruft GET /status auf (Deadline: eine Sekunde — der Daemon ist // lokal und muss antworten, sonst ist das selbst ein Testergebnis). func getStatus(t *testing.T, d *daemonProc) statusBody { t.Helper() client := &http.Client{Timeout: time.Second} resp, err := client.Get(d.baseURL + "/status") if err != nil { t.Fatalf("GET /status: %v — stderr:\n%s", err, d.stderr.String()) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { t.Fatalf("GET /status: Status %d", resp.StatusCode) } var st statusBody if err := json.NewDecoder(resp.Body).Decode(&st); err != nil { t.Fatalf("GET /status dekodieren: %v", err) } return st } // waitStatusIdle pollt, bis weder ein Item läuft noch wartet. func waitStatusIdle(t *testing.T, d *daemonProc, timeout time.Duration) { t.Helper() deadline := time.Now().Add(timeout) for { st := getStatus(t, d) if st.Current == nil && st.Queue == 0 { return } if time.Now().After(deadline) { t.Fatalf("Queue wird nicht idle: %+v — stderr:\n%s", st, d.stderr.String()) } time.Sleep(25 * time.Millisecond) } } // speak enqueue per rohem HTTP (Prozessgrenze ohne CLI). func speak(t *testing.T, d *daemonProc, body string) (int, map[string]any) { t.Helper() resp, err := http.Post(d.baseURL+"/speak", "application/json", strings.NewReader(body)) if err != nil { t.Fatalf("POST /speak: %v", err) } defer resp.Body.Close() var m map[string]any if err := json.NewDecoder(resp.Body).Decode(&m); err != nil { t.Fatalf("POST /speak Antwort dekodieren: %v", err) } return resp.StatusCode, m } // postPath macht einen leeren POST auf einen Daemon-Endpunkt (/mute, // /unmute, /cancel) und dekodiert die JSON-Antwort. func postPath(t *testing.T, d *daemonProc, path string) (int, map[string]any) { t.Helper() resp, err := http.Post(d.baseURL+path, "application/json", strings.NewReader("{}")) if err != nil { t.Fatalf("POST %s: %v", path, err) } defer resp.Body.Close() var m map[string]any if err := json.NewDecoder(resp.Body).Decode(&m); err != nil { t.Fatalf("POST %s Antwort dekodieren: %v", path, err) } return resp.StatusCode, m }