1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
package api
import (
"errors"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/thepeterstone/claudomator/internal/storage"
)
// mockLogStore implements logStore for testing.
type mockLogStore struct {
fn func(id string) (*storage.Execution, error)
}
func (m *mockLogStore) GetExecution(id string) (*storage.Execution, error) {
return m.fn(id)
}
func logsMux(srv *Server) *http.ServeMux {
mux := http.NewServeMux()
mux.HandleFunc("GET /api/executions/{id}/logs", srv.handleStreamLogs)
return mux
}
// TestHandleStreamLogs_NotFound verifies that an unknown execution ID yields 404.
func TestHandleStreamLogs_NotFound(t *testing.T) {
srv := &Server{
logStore: &mockLogStore{fn: func(id string) (*storage.Execution, error) {
return nil, errors.New("not found")
}},
}
req := httptest.NewRequest("GET", "/api/executions/nonexistent/logs", nil)
w := httptest.NewRecorder()
logsMux(srv).ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("status: want 404, got %d; body: %s", w.Code, w.Body.String())
}
}
// TestHandleStreamLogs_TerminalState_EmitsEventsAndDone verifies that a COMPLETED
// execution with a populated stdout.log streams SSE events and terminates with a done event.
func TestHandleStreamLogs_TerminalState_EmitsEventsAndDone(t *testing.T) {
dir := t.TempDir()
logPath := filepath.Join(dir, "stdout.log")
lines := strings.Join([]string{
`{"type":"assistant","message":{"content":[{"type":"text","text":"Hello world"}]}}`,
`{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Bash","input":{"command":"ls"}}]}}`,
`{"type":"result","cost_usd":0.0042}`,
}, "\n") + "\n"
if err := os.WriteFile(logPath, []byte(lines), 0600); err != nil {
t.Fatal(err)
}
exec := &storage.Execution{
ID: "exec-terminal-1",
TaskID: "task-terminal-1",
StartTime: time.Now(),
Status: "COMPLETED",
StdoutPath: logPath,
}
srv := &Server{
logStore: &mockLogStore{fn: func(id string) (*storage.Execution, error) {
return exec, nil
}},
}
req := httptest.NewRequest("GET", "/api/executions/exec-terminal-1/logs", nil)
w := httptest.NewRecorder()
logsMux(srv).ServeHTTP(w, req)
if ct := w.Header().Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
t.Errorf("Content-Type: want text/event-stream, got %q", ct)
}
body := w.Body.String()
if !strings.Contains(body, "data: ") {
t.Error("expected at least one SSE 'data: ' event in body")
}
if !strings.Contains(body, "\n\n") {
t.Error("expected SSE double-newline event termination")
}
if !strings.HasSuffix(body, "event: done\ndata: {}\n\n") {
t.Errorf("body does not end with done event; got:\n%s", body)
}
}
// TestHandleStreamLogs_EmptyLog verifies that a COMPLETED execution with no stdout path
// responds with only the done sentinel event.
func TestHandleStreamLogs_EmptyLog(t *testing.T) {
exec := &storage.Execution{
ID: "exec-empty-1",
TaskID: "task-empty-1",
StartTime: time.Now(),
Status: "COMPLETED",
// StdoutPath intentionally empty
}
srv := &Server{
logStore: &mockLogStore{fn: func(id string) (*storage.Execution, error) {
return exec, nil
}},
}
req := httptest.NewRequest("GET", "/api/executions/exec-empty-1/logs", nil)
w := httptest.NewRecorder()
logsMux(srv).ServeHTTP(w, req)
body := w.Body.String()
if body != "event: done\ndata: {}\n\n" {
t.Errorf("want only done event, got:\n%s", body)
}
}
// TestHandleStreamLogs_RunningState_LiveTail verifies that a RUNNING execution streams
// initial log content and emits a done event once it transitions to a terminal state.
func TestHandleStreamLogs_RunningState_LiveTail(t *testing.T) {
dir := t.TempDir()
logPath := filepath.Join(dir, "stdout.log")
logLines := strings.Join([]string{
`{"type":"assistant","message":{"content":[{"type":"text","text":"Working..."}]}}`,
`{"type":"result","cost_usd":0.001}`,
}, "\n") + "\n"
if err := os.WriteFile(logPath, []byte(logLines), 0600); err != nil {
t.Fatal(err)
}
runningExec := &storage.Execution{
ID: "exec-running-1",
TaskID: "task-running-1",
StartTime: time.Now(),
Status: "RUNNING",
StdoutPath: logPath,
}
// callCount tracks how many times GetExecution has been called.
// Call 1: initial fetch in handleStreamLogs → RUNNING
// Call 2+: poll in tailRunningExecution → COMPLETED
var callCount atomic.Int32
mock := &mockLogStore{fn: func(id string) (*storage.Execution, error) {
n := callCount.Add(1)
if n <= 1 {
return runningExec, nil
}
completed := *runningExec
completed.Status = "COMPLETED"
return &completed, nil
}}
srv := &Server{logStore: mock}
req := httptest.NewRequest("GET", "/api/executions/exec-running-1/logs", nil)
w := httptest.NewRecorder()
logsMux(srv).ServeHTTP(w, req)
body := w.Body.String()
if !strings.Contains(body, `"Working..."`) {
t.Errorf("expected initial text event in body; got:\n%s", body)
}
if !strings.Contains(body, `"type":"cost"`) {
t.Errorf("expected cost event in body; got:\n%s", body)
}
if !strings.HasSuffix(body, "event: done\ndata: {}\n\n") {
t.Errorf("body does not end with done event; got:\n%s", body)
}
}
|
