// Package test contains the process-boundary tests of talk 2.0 — the primary // test seam according to the spec: talkd and talk run as real processes // (built from this repo), driven via HTTP and the CLI, with two injected // fakes: an ElevenLabs fake endpoint (httptest server with audio bytes) // and audio.sink=null. Only externally observable behaviour is tested: // HTTP responses, CLI exit codes/output, fake hits (order and timing), // files in the config dir. Tests NEVER bind the default port 7800 and never // depend on a real daemon — talkd gets a free port via TALK_PORT, CLI // processes the same one (or port 1 for "daemon dead"). package test import ( "bytes" "encoding/json" "fmt" "io" "net" "net/http" "net/http/httptest" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "testing" "time" ) // The binaries are built once per test run (the build cache makes it cheap) // and shared by all tests. var talkBin 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) } build := exec.Command("go", "build", "-o", filepath.Join(dir, "talk"), "./cmd/talk") build.Dir = repo if out, err := build.CombinedOutput(); err != nil { fmt.Fprintf(os.Stderr, "go build ./cmd/talk: %v\n%s", err, out) os.RemoveAll(dir) os.Exit(1) } talkBin = filepath.Join(dir, "talk") code := m.Run() os.RemoveAll(dir) os.Exit(code) } // syncBuffer is a concurrency-safe buffer (daemon stderr is read while the // process is running). 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 is an observed synthesis request at the fake provider. type fakeHit struct { Text string // text from the request body Start time.Time // arrival at the fake End time.Time // response sent or cancellation seen Completed bool // true: response written; false: client cancelled } // fakeProvider is an ElevenLabs fake (POST /v1/text-to-speech/): it // answers with a few bytes of "audio" and logs every request with arrival // and response time — from which order (text sequence) and serialization // (arrival of the next after the response of the previous) can be read. // delay can be switched at runtime to simulate blocking syntheses // (serialization, priority, cancel). Arrivals are logged separately and // immediately: they show that the daemon is BUSY with a synthesis (queue // can be filled deterministically) before the (delayed) response lands as // a hit. 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) cancelled 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 returns only the completed (not cancelled) hits. func completed(hits []fakeHit) []fakeHit { var out []fakeHit for _, h := range hits { if h.Completed { out = append(out, h) } } return out } // waitArrived polls until n synthesis requests have arrived — the daemon is // then demonstrably busy and the queue can be filled deterministically. 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("only %d of %d arrivals at the fake within 5 s", len(f.snapshot()), n) } // arrivedCount counts the synthesis requests that arrived (including // cancelled ones) — observation point for "exactly N play, the rest // parks/expires". func (f *fakeProvider) arrivedCount() int { f.mu.Lock() defer f.mu.Unlock() return len(f.arrived) } // assertNoFurtherArrivals checks that no further synthesis request arrives // within the deadline (parked/discarded announcements stay silent). 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 additional arrivals (want none) — hits: %+v", got-n, f.snapshot()) } } // waitHits polls until n requests have finished (cancellations count). 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("only %d of %d fake hits within 5 s (hits: %+v)", len(f.snapshot()), n, f.snapshot()) return nil } // waitCompleted polls until n requests have finished SUCCESSFULLY // (cancelled syntheses do not count). 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("only %d of %d completed hits within 5 s (all: %+v)", len(completed(f.snapshot())), n, f.snapshot()) return nil } // assertNoHits checks that no request arrived within the deadline. func (f *fakeProvider) assertNoHits(t *testing.T, wait time.Duration) { t.Helper() time.Sleep(wait) if hits := f.snapshot(); len(hits) > 0 { t.Fatalf("unexpected fake hits: %+v", hits) } } // daemonProc is a running talkd process. type daemonProc struct { cmd *exec.Cmd port int stderr *syncBuffer baseURL string } // startDaemon starts `talk daemon run` as a real process (free port via // TALK_PORT, never the default port) and waits until GET /status answers. // Stopped via t.Cleanup: SIGINT, kill after a deadline. 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(talkBin, "daemon", "run") cmd.Env = env cmd.Stderr = stderr if err := cmd.Start(); err != nil { t.Fatalf("starting talk daemon run: %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("talk daemon does not answer on %s — stderr:\n%s", d.baseURL+"/status", stderr.String()) return nil } // stop shuts the daemon process down (SIGINT like a human stop; kill // fallback after 3 s) and waits for it to end. 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 builds the process environment for talk processes (CLI and // daemon): the inherited environment without foreign TALK_* variables // (isolation from keys/ports of the outside world), isolated HOME/ // XDG_CONFIG_HOME (the config dir is thus /talk), audio.sink=null and // the fake provider as ElevenLabs endpoint. port=1 means "guaranteed dead // daemon" (connection is refused immediately). // // TALK_RATE_LIMIT=0s is the base default of the process-boundary tests: the // Task-04 semantics tests (serialization, priority, …) enqueue in quick // succession and expect immediate playback — the law (default 30 s) would // park them. The law tests in ratelimit_test.go deliberately override the // value via 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": "process-test-key", "TALK_ELEVENLABS_ENDPOINT": fakeURL, "TALK_EDGE_ENDPOINT": "ws://127.0.0.1:1/", // the chain must never fall back to real third-party endpoints "TALK_PORT": strconv.Itoa(port), "TALK_RATE_LIMIT": "0s", // law off until a test turns it on (see above) "TALK_AUTOSTART": "false", // no test may leave a daemon behind unnoticed; autostart_test.go turns it on deliberately "TALK_NOTIFY": "false", // no desktop notifications on the developer's machine; notify_dbus_test.go turns it on against a private bus "DBUS_SESSION_BUS_ADDRESS": "unix:path=/nonexistent/talk-test-bus", // never the developer's real session bus (toast tests observe notify-send) } for k, v := range extra { set[k] = v } for k, v := range set { env = append(env, k+"="+v) } return env } // freePort obtains a free loopback port (bind-and-close; the daemon rebinds // it right afterwards). func freePort(t *testing.T) int { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("finding a free port: %v", err) } defer ln.Close() return ln.Addr().(*net.TCPAddr).Port } // configDir is the config directory under the isolated root (Linux path // logic: XDG_CONFIG_HOME/ → /talk). func configDir(root string) string { return filepath.Join(root, "talk") } // talkResult is the observed result of a talk process. type talkResult struct { code int stdout string stderr string } // runTalk runs the talk CLI as a real process. 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("running talk: %v (stderr: %s)", err, errBuf.String()) } return talkResult{code: code, stdout: out.String(), stderr: errBuf.String()} } // statusBody is the decoded GET /status response. 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"` MCP bool `json:"mcp"` Cors []string `json:"cors"` } // getStatus calls GET /status (deadline: one second — the daemon is local // and must answer, otherwise that is a test result in itself). 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("decoding GET /status: %v", err) } return st } // waitStatusIdle polls until neither an item is running nor waiting. 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 does not become idle: %+v — stderr:\n%s", st, d.stderr.String()) } time.Sleep(25 * time.Millisecond) } } // speak enqueues via raw HTTP (process boundary without the 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("decoding POST /speak response: %v", err) } return resp.StatusCode, m } // postPath makes an empty POST to a daemon endpoint (/mute, /unmute, // /cancel) and decodes the JSON response. 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("decoding POST %s response: %v", path, err) } return resp.StatusCode, m } // rawResponse is a fully read HTTP response (status, header, body as a // string) — the observation form of the raw CORS/MCP tests. type rawResponse struct { status int header http.Header body string } // doRequest makes a raw HTTP request with arbitrary headers and an optional // body and reads the response completely (CORS preflights and MCP JSON-RPC // payloads are literal, no client SDK in between). func doRequest(t *testing.T, method, url string, headers map[string]string, body string) rawResponse { t.Helper() var rd io.Reader if body != "" { rd = strings.NewReader(body) } req, err := http.NewRequest(method, url, rd) if err != nil { t.Fatalf("building %s %s: %v", method, url, err) } for k, v := range headers { req.Header.Set(k, v) } resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("%s %s: %v", method, url, err) } defer resp.Body.Close() raw, err := io.ReadAll(resp.Body) if err != nil { t.Fatalf("reading %s %s response: %v", method, url, err) } return rawResponse{status: resp.StatusCode, header: resp.Header, body: string(raw)} } // preflight makes an OPTIONS preflight against /speak (the way a browser // does before a real POST). func preflight(t *testing.T, d *daemonProc, origin string) rawResponse { t.Helper() return doRequest(t, http.MethodOptions, d.baseURL+"/speak", map[string]string{ "Origin": origin, "Access-Control-Request-Method": http.MethodPost, }, "") } // mcpPost POSTs a literal JSON-RPC payload to /mcp. func mcpPost(t *testing.T, d *daemonProc, payload string) rawResponse { t.Helper() return doRequest(t, http.MethodPost, d.baseURL+"/mcp", map[string]string{ "Content-Type": "application/json", }, payload) } // mcpDecode decodes a JSON-RPC response into a generic map (for field // access in the literal payload tests). func mcpDecode(t *testing.T, raw rawResponse) map[string]any { t.Helper() var m map[string]any if err := json.Unmarshal([]byte(raw.body), &m); err != nil { t.Fatalf("decoding JSON-RPC response: %v — body: %s", err, raw.body) } return m }