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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
|
package jsonrpc_test
import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"strings"
"testing"
"time"
"codeberg.org/turbo-editors/turbo-core/jsonrpc"
)
// newlineFraming marks messages off with a newline, which is what the Agent
// Client Protocol does. It is the simplest framing there is, so the tests are
// about the JSON-RPC layer rather than about parsing headers.
type newlineFraming struct{}
func (newlineFraming) ReadMessage(r *bufio.Reader) ([]byte, error) {
line, err := r.ReadBytes('\n')
if err != nil {
return nil, err
}
return line, nil
}
func (newlineFraming) WriteMessage(w io.Writer, body []byte) error {
_, err := fmt.Fprintf(w, "%s\n", body)
return err
}
// frame is one message the peer read, or the reason it could not.
type frame struct {
msg map[string]any
err error
}
// peer is the far end of a connection, driven from a test.
//
// It is deliberately not built on Conn: a peer sharing the code under test
// could not catch that code writing a malformed frame.
//
// It reads continuously into a buffered channel rather than on demand, because
// net.Pipe is synchronous — a write blocks until somebody reads. With on-demand
// reads, a Reply made from the test goroutine deadlocks against the read that
// was going to collect it. Reading continuously is also what a real client
// does.
type peer struct {
t *testing.T
stream net.Conn
frames chan frame
}
// newPair returns a Conn and the raw peer at the other end of it.
func newPair(t *testing.T, onNotify jsonrpc.NotificationFunc, onRequest jsonrpc.RequestFunc) (*jsonrpc.Conn, *peer) {
t.Helper()
ours, theirs := net.Pipe()
conn := jsonrpc.NewConn(ours, newlineFraming{}, onNotify, onRequest)
go func() { _ = conn.Run() }()
p := &peer{t: t, stream: theirs, frames: make(chan frame, 32)}
go p.readLoop()
t.Cleanup(func() { _ = conn.Close(); _ = theirs.Close() })
return conn, p
}
// readLoop decodes every line the connection sends until it ends.
//
// It never calls t.Fatalf: that is not allowed outside the test goroutine, and
// doing it here hangs the run instead of failing it. Failures travel back as
// values instead.
func (p *peer) readLoop() {
reader := bufio.NewReader(p.stream)
for {
line, err := reader.ReadBytes('\n')
if err != nil {
close(p.frames)
return
}
var msg map[string]any
if err := json.Unmarshal(line, &msg); err != nil {
p.frames <- frame{err: fmt.Errorf("the connection wrote %q, which is not JSON: %w", line, err)}
continue
}
p.frames <- frame{msg: msg}
}
}
// send writes one raw line to the connection under test.
func (p *peer) send(line string) {
p.t.Helper()
_ = p.stream.SetWriteDeadline(time.Now().Add(2 * time.Second))
if _, err := fmt.Fprintf(p.stream, "%s\n", line); err != nil {
p.t.Fatalf("writing to the connection: %v", err)
}
}
// read returns the next message the connection sent, failing the test if none
// arrives.
func (p *peer) read() map[string]any {
p.t.Helper()
select {
case got, open := <-p.frames:
if !open {
p.t.Fatal("the connection ended without sending anything more")
}
if got.err != nil {
p.t.Fatal(got.err)
}
return got.msg
case <-time.After(2 * time.Second):
p.t.Fatal("the connection sent nothing")
return nil
}
}
// readsNothing fails the test if the connection sends anything within a short
// window. It is how "this must not be answered" is stated.
func (p *peer) readsNothing(what string) {
p.t.Helper()
select {
case got := <-p.frames:
p.t.Fatalf("%s: the connection sent %v", what, got.msg)
case <-time.After(150 * time.Millisecond):
}
}
func TestACallGetsItsAnswer(t *testing.T) {
conn, peer := newPair(t, nil, nil)
answered := make(chan error, 1)
var result struct{ Name string }
go func() { answered <- conn.Call(t.Context(), "greet", map[string]string{"who": "world"}, &result) }()
request := peer.read()
if request["method"] != "greet" {
t.Fatalf("the connection sent %v, want a greet request", request)
}
if request["jsonrpc"] != jsonrpc.Version {
t.Errorf("the request says jsonrpc %v, want %q", request["jsonrpc"], jsonrpc.Version)
}
peer.send(fmt.Sprintf(`{"jsonrpc":"2.0","id":%v,"result":{"Name":"hello"}}`, request["id"]))
if err := <-answered; err != nil {
t.Fatalf("Call() error = %v", err)
}
if result.Name != "hello" {
t.Errorf("the result is %q, want %q", result.Name, "hello")
}
}
func TestAPeersRequestIdDoesNotStealAnAnswerMeantForOurOwnCall(t *testing.T) {
// Both sides number their requests from one, and an agent really does send
// `id: 1` while a call of ours with the same id is in flight — that is what
// the Agent Client Protocol's agents do. The two directions are separate
// id spaces, and a connection that shared one map would hand the peer's
// *request* to the caller waiting for an answer, or answer the request
// with the caller's result.
asked := make(chan string, 1)
onRequest := func(req *jsonrpc.Request) {
asked <- req.Method
req.Reply(map[string]string{"outcome": "allowed"}, nil)
}
conn, peer := newPair(t, nil, onRequest)
answered := make(chan error, 1)
var result struct{ Name string }
go func() { answered <- conn.Call(t.Context(), "ours", nil, &result) }()
ourRequest := peer.read()
if fmt.Sprint(ourRequest["id"]) != "1" {
t.Fatalf("our first call took id %v, want 1 — the rest of this test assumes it", ourRequest["id"])
}
// The peer now asks a question carrying the same id as our call in flight.
peer.send(`{"jsonrpc":"2.0","id":1,"method":"theirs","params":{}}`)
select {
case method := <-asked:
if method != "theirs" {
t.Errorf("the handler was given %q, want %q", method, "theirs")
}
case <-time.After(2 * time.Second):
t.Fatal("the peer's request never reached the handler")
}
reply := peer.read()
if fmt.Sprint(reply["id"]) != "1" {
t.Errorf("the answer carries id %v, want the peer's own 1", reply["id"])
}
if reply["result"] == nil {
t.Errorf("the peer's request was answered with %v, want a result", reply)
}
// Our own call must still be waiting, and must still get its own answer.
select {
case err := <-answered:
t.Fatalf("our call finished early with %v; the peer's request stole its answer", err)
case <-time.After(50 * time.Millisecond):
}
peer.send(`{"jsonrpc":"2.0","id":1,"result":{"Name":"ours"}}`)
if err := <-answered; err != nil {
t.Fatalf("Call() error = %v", err)
}
if result.Name != "ours" {
t.Errorf("our call got %q, want %q", result.Name, "ours")
}
}
func TestARequestMayBeAnsweredLongAfterItArrived(t *testing.T) {
// This is the whole reason this package exists apart from lsp: a permission
// dialog is answered by a person, on another goroutine, several turns of an
// event loop later. A handler that had to return its answer could not wait.
held := make(chan *jsonrpc.Request, 1)
_, peer := newPair(t, nil, func(req *jsonrpc.Request) { held <- req })
peer.send(`{"jsonrpc":"2.0","id":7,"method":"session/request_permission","params":{}}`)
request := <-held
if request.Method != "session/request_permission" {
t.Fatalf("the handler was given %q", request.Method)
}
// Nothing may be written until the answer is given.
peer.readsNothing("answered before the handler did")
// From another goroutine, as the event loop's dialog really does.
go request.Reply(map[string]any{"outcome": map[string]string{"outcome": "selected"}}, nil)
reply := peer.read()
if fmt.Sprint(reply["id"]) != "7" {
t.Errorf("the answer carries id %v, want 7", reply["id"])
}
if reply["result"] == nil {
t.Errorf("the answer is %v, want a result", reply)
}
}
func TestOnlyTheFirstReplyToARequestIsSent(t *testing.T) {
// A window closing tears down a pending permission that may already have
// been answered. Two responses carrying one id would desynchronise the
// peer, which matches answers to requests by exactly that id.
held := make(chan *jsonrpc.Request, 1)
_, peer := newPair(t, nil, func(req *jsonrpc.Request) { held <- req })
peer.send(`{"jsonrpc":"2.0","id":3,"method":"ask","params":{}}`)
request := <-held
go request.Reply("first", nil)
first := peer.read()
// Only now, with the first answer known to be on the wire, can the later
// ones be tried: Reply holds the once until its write has finished.
request.Reply("second", nil)
request.Reply(nil, errors.New("third"))
if first["result"] != "first" {
t.Errorf("the answer is %v, want the first one", first)
}
// A second frame must never arrive.
peer.readsNothing("a second answer was sent")
}
func TestAnErrorFromAHandlerReachesThePeerAsAnErrorObject(t *testing.T) {
_, peer := newPair(t, nil, func(req *jsonrpc.Request) {
req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeInvalidParams, Message: "no uri"})
})
peer.send(`{"jsonrpc":"2.0","id":2,"method":"fs/read_text_file","params":{}}`)
reply := peer.read()
failure, ok := reply["error"].(map[string]any)
if !ok {
t.Fatalf("the answer is %v, want an error object", reply)
}
if failure["code"] != float64(jsonrpc.CodeInvalidParams) {
t.Errorf("the code is %v, want %d", failure["code"], jsonrpc.CodeInvalidParams)
}
if failure["message"] != "no uri" {
t.Errorf("the message is %v, want %q", failure["message"], "no uri")
}
}
func TestAPlainErrorFromAHandlerBecomesAnInternalError(t *testing.T) {
_, peer := newPair(t, nil, func(req *jsonrpc.Request) {
req.Reply(nil, errors.New("the file is a directory"))
})
peer.send(`{"jsonrpc":"2.0","id":2,"method":"fs/read_text_file","params":{}}`)
failure, ok := peer.read()["error"].(map[string]any)
if !ok {
t.Fatal("want an error object")
}
if failure["code"] != float64(jsonrpc.CodeInternalError) {
t.Errorf("the code is %v, want %d", failure["code"], jsonrpc.CodeInternalError)
}
if !strings.Contains(fmt.Sprint(failure["message"]), "the file is a directory") {
t.Errorf("the message is %v, want the handler's own words in it", failure["message"])
}
}
func TestARequestWithNoHandlerIsAnsweredMethodNotFound(t *testing.T) {
// Silence would leave the peer waiting for ever. A peer that asks for
// something this client does not do has to be told so.
_, peer := newPair(t, nil, nil)
peer.send(`{"jsonrpc":"2.0","id":4,"method":"terminal/create","params":{}}`)
failure, ok := peer.read()["error"].(map[string]any)
if !ok {
t.Fatal("want an error object")
}
if failure["code"] != float64(jsonrpc.CodeMethodNotFound) {
t.Errorf("the code is %v, want %d", failure["code"], jsonrpc.CodeMethodNotFound)
}
if failure["message"] != "terminal/create" {
t.Errorf("the message is %v, want the method's name", failure["message"])
}
}
func TestANotificationReachesTheHandlerAndIsNeverAnswered(t *testing.T) {
// A notification is exactly a message with no id, and answering one would
// send a response the peer has nothing to match it to.
arrived := make(chan string, 1)
_, peer := newPair(t, func(method string, _ json.RawMessage) { arrived <- method }, nil)
peer.send(`{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"s"}}`)
select {
case method := <-arrived:
if method != "session/update" {
t.Errorf("the handler was given %q", method)
}
case <-time.After(2 * time.Second):
t.Fatal("the notification never arrived")
}
peer.readsNothing("the notification was answered")
}
func TestNotifySendsNoId(t *testing.T) {
conn, peer := newPair(t, nil, nil)
go func() { _ = conn.Notify("session/cancel", map[string]string{"sessionId": "s"}) }()
sent := peer.read()
if _, carries := sent["id"]; carries {
t.Errorf("the notification carries an id: %v", sent)
}
if sent["method"] != "session/cancel" {
t.Errorf("the notification is %v", sent)
}
}
func TestACallFailsWhenTheConnectionEnds(t *testing.T) {
// A peer that dies mid-request must not leave the caller blocked for ever.
conn, peer := newPair(t, nil, nil)
answered := make(chan error, 1)
go func() { answered <- conn.Call(context.Background(), "greet", nil, nil) }()
peer.read() // wait until the request is really on the wire
_ = peer.stream.Close()
select {
case err := <-answered:
if !errors.Is(err, jsonrpc.ErrClosed) {
t.Errorf("Call() error = %v, want ErrClosed", err)
}
case <-time.After(2 * time.Second):
t.Fatal("the call never came back")
}
}
func TestACallEndsWithItsContext(t *testing.T) {
conn, peer := newPair(t, nil, nil)
ctx, cancel := context.WithCancel(context.Background())
answered := make(chan error, 1)
go func() { answered <- conn.Call(ctx, "slow", nil, nil) }()
peer.read()
cancel()
select {
case err := <-answered:
if !errors.Is(err, context.Canceled) {
t.Errorf("Call() error = %v, want context.Canceled", err)
}
case <-time.After(2 * time.Second):
t.Fatal("the call ignored its context")
}
}
func TestAResponseToAnIdWeNeverSentIsIgnored(t *testing.T) {
// The connection must not panic or die on one; an agent replaying a
// recorded session can easily send one.
conn, peer := newPair(t, nil, nil)
peer.send(`{"jsonrpc":"2.0","id":999,"result":{}}`)
// Still usable afterwards, which is the whole point.
answered := make(chan error, 1)
go func() { answered <- conn.Call(t.Context(), "greet", nil, nil) }()
request := peer.read()
peer.send(fmt.Sprintf(`{"jsonrpc":"2.0","id":%v,"result":null}`, request["id"]))
if err := <-answered; err != nil {
t.Errorf("the connection stopped working after a stray response: %v", err)
}
}
func TestCallingOnAClosedConnectionFails(t *testing.T) {
conn, _ := newPair(t, nil, nil)
if err := conn.Close(); err != nil {
t.Fatalf("Close() error = %v", err)
}
if err := conn.Call(t.Context(), "greet", nil, nil); !errors.Is(err, jsonrpc.ErrClosed) {
t.Errorf("Call() error = %v, want ErrClosed", err)
}
}
|