package handlers import ( "encoding/json" "net/http/httptest" "strings" "testing" "time" ) // recorderFlusher wraps httptest.ResponseRecorder to satisfy http.Flusher. type recorderFlusher struct { *httptest.ResponseRecorder } func (rf *recorderFlusher) Flush() {} func newRecorderFlusher() *recorderFlusher { return &recorderFlusher{httptest.NewRecorder()} } func TestStreamManager_SubscribeAndSend(t *testing.T) { sm := newStreamManager() ch := sm.Subscribe("stream-1") defer sm.Unsubscribe("stream-1", ch) // Send event sm.Send("stream-1", "output", map[string]any{"line": "hello"}) select { case evt := <-ch: if evt.Type != "output" { t.Errorf("event type = %q, want %q", evt.Type, "output") } if evt.Data["line"] != "hello" { t.Errorf("event data = %v, want line=hello", evt.Data) } case <-time.After(time.Second): t.Fatal("timed out waiting for event") } } func TestStreamManager_MultipleSubscribers(t *testing.T) { sm := newStreamManager() ch1 := sm.Subscribe("stream-1") ch2 := sm.Subscribe("stream-1") defer sm.Unsubscribe("stream-1", ch1) defer sm.Unsubscribe("stream-1", ch2) sm.Send("stream-1", "test", map[string]any{"value": 1}) // Both should receive for i, ch := range []chan streamEvent{ch1, ch2} { select { case evt := <-ch: if evt.Type != "test" { t.Errorf("subscriber %d: event type = %q, want %q", i, evt.Type, "test") } case <-time.After(time.Second): t.Fatalf("subscriber %d: timed out", i) } } } func TestStreamManager_Close(t *testing.T) { sm := newStreamManager() ch := sm.Subscribe("stream-1") sm.Close("stream-1") // Channel should be closed _, ok := <-ch if ok { t.Error("channel should be closed after Close()") } } func TestStreamManager_SendToNonexistentStream(t *testing.T) { sm := newStreamManager() // Should not panic sm.Send("nonexistent", "test", map[string]any{}) } func TestStreamManager_Unsubscribe(t *testing.T) { sm := newStreamManager() ch1 := sm.Subscribe("stream-1") ch2 := sm.Subscribe("stream-1") sm.Unsubscribe("stream-1", ch1) // ch1 should be closed _, ok := <-ch1 if ok { t.Error("ch1 should be closed after Unsubscribe") } // ch2 should still receive sm.Send("stream-1", "test", map[string]any{}) select { case evt := <-ch2: if evt.Type != "test" { t.Errorf("event type = %q, want %q", evt.Type, "test") } case <-time.After(time.Second): t.Fatal("ch2 timed out") } sm.Unsubscribe("stream-1", ch2) } func TestWriteSSE(t *testing.T) { rf := newRecorderFlusher() writeSSE(rf.ResponseRecorder, rf, "output", map[string]any{"line": "hello"}) body := rf.Body.String() if !strings.Contains(body, "event: output\n") { t.Errorf("missing event line in SSE output: %s", body) } if !strings.Contains(body, "data: ") { t.Errorf("missing data line in SSE output: %s", body) } // Should not have id line if strings.Contains(body, "id: ") { t.Errorf("should not have id line without ID: %s", body) } // Verify data is valid JSON lines := strings.Split(body, "\n") for _, line := range lines { if strings.HasPrefix(line, "data: ") { jsonStr := strings.TrimPrefix(line, "data: ") var parsed map[string]any if err := json.Unmarshal([]byte(jsonStr), &parsed); err != nil { t.Errorf("data is not valid JSON: %v", err) } if parsed["line"] != "hello" { t.Errorf("data[line] = %v, want hello", parsed["line"]) } } } } func TestWriteSSEWithID(t *testing.T) { rf := newRecorderFlusher() writeSSEWithID(rf.ResponseRecorder, rf, "evt-123", "complete", map[string]any{"exit_code": 0}) body := rf.Body.String() if !strings.Contains(body, "id: evt-123\n") { t.Errorf("missing id line in SSE output: %s", body) } if !strings.Contains(body, "event: complete\n") { t.Errorf("missing event line in SSE output: %s", body) } } func TestStreamManager_FullChannel(t *testing.T) { sm := newStreamManager() ch := sm.Subscribe("stream-1") // Fill the channel (buffer size is 100) for i := 0; i < 100; i++ { sm.Send("stream-1", "test", map[string]any{"i": i}) } // Next send should not block (dropped) done := make(chan struct{}) go func() { sm.Send("stream-1", "test", map[string]any{"i": 100}) close(done) }() select { case <-done: // Good - did not block case <-time.After(time.Second): t.Fatal("Send blocked on full channel") } sm.Unsubscribe("stream-1", ch) }