diff --git a/README.md b/README.md index a168d16..492a16f 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ One harness works with: | **`nexssp/kernel` actions** | Auto-mount actions, context bridge for tenant/user, route discovery | | **Standard `http.Handler`** | Works with Chi, Gin, Echo, stdlib `http.ServeMux` — zero kernel required | | **Live E2E URLs** | Black-box staging/prod tests using the exact same fluent DSL | +| **JSON-RPC / Stdio transports** | Test line-delimited protocols (MCP stdio, custom JSON-RPC) with a real client | ### Why testkit outperforms ordinary testing setups @@ -26,6 +27,7 @@ One harness works with: - 📡 **SSE capture** — test realtime event streams natively - 💣 **Chaos injection** — latency, 503s, and panics with a few lines - 🧩 **Deterministic retry scripting** — unit-test circuit breakers without fakes +- 🔌 **JSON-RPC / Stdio testing** — test non-HTTP transports with the same real-client style ### The developer experience @@ -84,6 +86,7 @@ go get github.com/nexssp/testkit@latest 9. [Concurrency & Thundering-Herd Barriers](#9-concurrency--thundering-herd-barriers) 10. [In-Process Load Testing & P99 Latency Profiling](#10-in-process-load-testing--p99-latency-profiling) 11. [Chaos & Fault Injection](#11-chaos--fault-injection) +12. [Testing JSON-RPC / Stdio Transports](#12-testing-json-rpc--stdio-transports) --- @@ -317,6 +320,31 @@ func TestRealtimeTelemetry_SSE(t *testing.T) { } ``` +For MCP-style SSE handshakes, `Endpoint()` extracts the `event: endpoint` data, and `WaitForData()` blocks until an event contains a substring: + +```go +stream := suite.ListenSSE("/mcp/sse") +endpoint := stream.Endpoint(t, 2*time.Second) // "/mcp/message?sessionId=..." + +// POST to endpoint... + +msg := stream.WaitForData(t, "result", 2*time.Second) +``` + +For A2A / MCP Streamable HTTP (POST + SSE), use `ListenSSEWithRequest`: + +```go +req := httptest.NewRequest(http.MethodPost, "/endpoint", + strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"message/send"}`)) +req.Header.Set("Content-Type", "application/json") +req.Header.Set("Accept", "text/event-stream") + +stream := suite.ListenSSEWithRequest(t, req) +defer stream.Close() + +evt := stream.WaitFor(t, "message", 2*time.Second) +``` + --- ## 8. Deterministic Scripting & Hook Event Recording @@ -492,6 +520,65 @@ func TestResilience_UnderNetworkChaos(t *testing.T) { --- +## 12. Testing JSON-RPC / Stdio Transports + +Use `testkit/rpc` for line-delimited JSON-RPC transports such as MCP stdio. It dials an in-memory `net.Pipe` and speaks JSON-RPC 2.0 exactly like a real client. + +```go +import ( + "context" + "io" + "testing" + + "github.com/nexssp/testkit/rpc" +) + +func TestMCP_Stdio(t *testing.T) { + client := rpc.DialJSONRPC(t, func(ctx context.Context, in io.Reader, out io.Writer) error { + return mcpServer.Serve(ctx, in, out) + }) + + resp := client.Call("tools/list", nil, 1) + + var data struct { + Tools []struct { + Name string `json:"name"` + } `json:"tools"` + } + resp.BindResult(t, &data) + + // Assert with standard Go testing + if len(data.Tools) != 1 { + t.Fatalf("expected 1 tool, got %d", len(data.Tools)) + } +} +``` + +### `rpc.Client` API + +```go +client := rpc.DialJSONRPC(t, serveFunc) + +// Request / response +resp := client.Call("tools/list", nil, 1) + +// Notification (no response) +_ = client.Notify("notifications/initialized", nil) + +// Raw payload line (e.g. parse-error tests) +resp = client.CallRaw(`{"jsonrpc":"2.0","id":1,"method":"ping"}`) + +// Typed result binding +resp.BindResult(t, &myStruct) + +// JSON-RPC error object +if resp.Error != nil { + t.Fatalf("RPC error: %+v", resp.Error) +} +``` + +--- + ## License Apache License 2.0. See [LICENSE](LICENSE) for details. diff --git a/rpc/jsonrpc.go b/rpc/jsonrpc.go new file mode 100644 index 0000000..7f105d6 --- /dev/null +++ b/rpc/jsonrpc.go @@ -0,0 +1,181 @@ +package rpc + +import ( + "context" + "encoding/json" + "errors" + "io" + "net" + "testing" +) + +// Request is a JSON-RPC 2.0 request object. +type Request struct { + JSONRPC string `json:"jsonrpc"` + ID any `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +// Response is a JSON-RPC 2.0 response object. +type Response struct { + JSONRPC string `json:"jsonrpc"` + ID any `json:"id,omitempty"` + Result any `json:"result,omitempty"` + Error *RPCError `json:"error,omitempty"` +} + +// RPCError represents a JSON-RPC error object. +type RPCError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +func (e *RPCError) Error() string { return e.Message } + +// Client is a JSON-RPC 2.0 test client connected to an in-memory pipe. +type Client struct { + t testing.TB + conn net.Conn + enc *json.Encoder + dec *json.Decoder + cancel context.CancelFunc + done chan struct{} +} + +// DialJSONRPC creates a client and runs serve in a goroutine over net.Pipe. +func DialJSONRPC(t testing.TB, serve func(ctx context.Context, in io.Reader, out io.Writer) error) *Client { + t.Helper() + + serverConn, clientConn := net.Pipe() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + + go func() { + defer close(done) + defer serverConn.Close() + + err := serve(ctx, serverConn, serverConn) + if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) { + t.Errorf("rpc: server error: %v", err) + } + }() + + c := &Client{ + t: t, + conn: clientConn, + enc: json.NewEncoder(clientConn), + dec: json.NewDecoder(clientConn), + cancel: cancel, + done: done, + } + + t.Cleanup(func() { + cancel() + _ = clientConn.Close() + <-done + }) + + return c +} + +// Call sends a JSON-RPC request and waits for the matching response. +func (c *Client) Call(method string, params any, id any) Response { + c.t.Helper() + + raw, err := marshalParams(params) + if err != nil { + c.t.Fatalf("rpc: marshal params failed: %v", err) + } + + req := Request{ + JSONRPC: "2.0", + ID: id, + Method: method, + Params: raw, + } + + if err := c.enc.Encode(req); err != nil { + c.t.Fatalf("rpc: write request failed: %v", err) + } + + var resp Response + if err := c.dec.Decode(&resp); err != nil { + c.t.Fatalf("rpc: read response failed: %v", err) + } + + return resp +} + +// CallRaw sends a raw payload line and reads a response. +func (c *Client) CallRaw(payload string) Response { + c.t.Helper() + + if _, err := io.WriteString(c.conn, payload+"\n"); err != nil { + c.t.Fatalf("rpc: write raw request failed: %v", err) + } + + var resp Response + if err := c.dec.Decode(&resp); err != nil { + c.t.Fatalf("rpc: read response failed: %v", err) + } + + return resp +} + +// Notify sends a JSON-RPC notification (no ID, no response). +func (c *Client) Notify(method string, params any) error { + raw, err := marshalParams(params) + if err != nil { + return err + } + + req := Request{ + JSONRPC: "2.0", + Method: method, + Params: raw, + } + + return c.enc.Encode(req) +} + +// Close closes the client connection. +func (c *Client) Close() error { + c.cancel() + err := c.conn.Close() + <-c.done + return err +} + +// BindResult unmarshals a successful result into v. +func (r *Response) BindResult(t testing.TB, v any) { + t.Helper() + + if r.Error != nil { + t.Fatalf("rpc: response contains error: %+v", r.Error) + } + if r.Result == nil { + t.Fatalf("rpc: response result is nil") + } + + data, err := json.Marshal(r.Result) + if err != nil { + t.Fatalf("rpc: marshal result failed: %v", err) + } + if err := json.Unmarshal(data, v); err != nil { + t.Fatalf("rpc: unmarshal result into %T failed: %v", v, err) + } +} + +func marshalParams(params any) (json.RawMessage, error) { + if params == nil { + return nil, nil + } + + data, err := json.Marshal(params) + if err != nil { + return nil, err + } + + return json.RawMessage(data), nil +} diff --git a/rpc/jsonrpc_test.go b/rpc/jsonrpc_test.go new file mode 100644 index 0000000..c6fc528 --- /dev/null +++ b/rpc/jsonrpc_test.go @@ -0,0 +1,108 @@ +package rpc_test + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "io" + "testing" + + "github.com/nexssp/testkit/rpc" +) + +func serveEcho(ctx context.Context, in io.Reader, out io.Writer) error { + dec := json.NewDecoder(in) + enc := json.NewEncoder(out) + + for { + var req rpc.Request + if err := dec.Decode(&req); err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + + if req.ID == nil { + continue // notification + } + + if err := enc.Encode(rpc.Response{ + JSONRPC: "2.0", + ID: req.ID, + Result: map[string]any{"method": req.Method}, + }); err != nil { + return err + } + } +} + +func TestDialJSONRPC_Call(t *testing.T) { + client := rpc.DialJSONRPC(t, serveEcho) + + resp := client.Call("tools/list", nil, 1) + if resp.Error != nil { + t.Fatalf("unexpected error: %+v", resp.Error) + } + + var result map[string]any + resp.BindResult(t, &result) + if result["method"] != "tools/list" { + t.Fatalf("unexpected result: %+v", result) + } +} + +func TestDialJSONRPC_Notify(t *testing.T) { + client := rpc.DialJSONRPC(t, serveEcho) + if err := client.Notify("notifications/initialized", nil); err != nil { + t.Fatalf("notify failed: %v", err) + } +} + +func TestDialJSONRPC_Error(t *testing.T) { + server := func(ctx context.Context, in io.Reader, out io.Writer) error { + dec := json.NewDecoder(in) + enc := json.NewEncoder(out) + + var req rpc.Request + if err := dec.Decode(&req); err != nil { + return err + } + + return enc.Encode(rpc.Response{ + JSONRPC: "2.0", + ID: req.ID, + Error: &rpc.RPCError{Code: -32601, Message: "method not found"}, + }) + } + + client := rpc.DialJSONRPC(t, server) + + resp := client.Call("nope", nil, 2) + if resp.Error == nil || resp.Error.Code != -32601 { + t.Fatalf("expected -32601 error, got %+v", resp.Error) + } +} + +func TestDialJSONRPC_CallRaw(t *testing.T) { + server := func(ctx context.Context, in io.Reader, out io.Writer) error { + br := bufio.NewReader(in) + if _, err := br.ReadBytes('\n'); err != nil { + return err + } + + enc := json.NewEncoder(out) + return enc.Encode(rpc.Response{ + JSONRPC: "2.0", + Error: &rpc.RPCError{Code: -32700, Message: "Parse error"}, + }) + } + + client := rpc.DialJSONRPC(t, server) + + resp := client.CallRaw("{invalid json}") + if resp.Error == nil || resp.Error.Code != -32700 { + t.Fatalf("expected parse error, got %+v", resp.Error) + } +} diff --git a/sse.go b/sse.go index 16fe8da..06b91c8 100644 --- a/sse.go +++ b/sse.go @@ -5,6 +5,7 @@ import ( "bytes" "context" "net/http" + "net/url" "strings" "sync" "testing" @@ -21,24 +22,79 @@ type StreamCapture struct { cancel context.CancelFunc } -// ListenSSE connects to an SSE endpoint, unblocking immediately on initial HTTP response. +// ListenSSE connects to an SSE endpoint using GET. func (s *Suite) ListenSSE(path string) *StreamCapture { s.T.Helper() + return s.listenSSE(s.T, func(ctx context.Context) *http.Request { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.baseURL+path, http.NoBody) + if err != nil { + s.T.Fatalf("ListenSSE: failed to build request: %v", err) + } + req.Header.Set("Accept", "text/event-stream") + return req + }) +} + +// ListenSSEWithRequest connects to an SSE endpoint using a custom HTTP request. +// This supports POST + SSE patterns such as MCP Streamable HTTP and A2A. +func (s *Suite) ListenSSEWithRequest(t testing.TB, req *http.Request) *StreamCapture { + t.Helper() + + return s.listenSSE(t, func(ctx context.Context) *http.Request { + cloned := req.Clone(ctx) + + // httptest.NewRequest creates a server-side request. + // http.Client.Do requires RequestURI to be empty. + cloned.RequestURI = "" + + if req.URL != nil { + u := *req.URL + cloned.URL = &u + } else { + cloned.URL = &url.URL{} + } + + s.setBaseURL(t, cloned) + + if cloned.Header.Get("Accept") == "" { + cloned.Header.Set("Accept", "text/event-stream") + } + + return cloned + }) +} + +func (s *Suite) setBaseURL(t testing.TB, req *http.Request) { + t.Helper() + + base, err := url.Parse(s.baseURL) + if err != nil { + t.Fatalf("ListenSSE: invalid base URL %q: %v", s.baseURL, err) + } + + req.URL.Scheme = base.Scheme + req.URL.Host = base.Host + + if req.URL.Path == "" { + req.URL.Path = "/" + } + if !strings.HasPrefix(req.URL.Path, "/") { + req.URL.Path = "/" + req.URL.Path + } +} + +func (s *Suite) listenSSE(t testing.TB, buildReq func(ctx context.Context) *http.Request) *StreamCapture { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) streamCap := &StreamCapture{ events: make(chan SSEEvent, 100), cancel: cancel, } - s.T.Cleanup(streamCap.Close) - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.baseURL+path, http.NoBody) - if err != nil { - cancel() - s.T.Fatalf("ListenSSE: failed to build request: %v", err) - } + t.Cleanup(streamCap.Close) - req.Header.Set("Accept", "text/event-stream") + req := buildReq(ctx) s.mu.RLock() for k, v := range s.headers { @@ -54,7 +110,6 @@ func (s *Suite) ListenSSE(path string) *StreamCapture { go func() { resp, err := s.client.Do(req) - // Close ready on ANY response arrival to prevent hangs on error status codes readyOnce.Do(func() { close(ready) }) if err != nil { @@ -98,7 +153,7 @@ func (s *Suite) ListenSSE(path string) *StreamCapture { case <-ready: case <-time.After(3 * time.Second): streamCap.Close() - s.T.Fatalf("ListenSSE: timed out waiting for SSE connection on %s", path) + t.Fatalf("ListenSSE: timed out waiting for SSE connection on %s", req.URL.Path) } return streamCap @@ -121,6 +176,30 @@ func (sc *StreamCapture) WaitFor(t testing.TB, eventName string, timeout time.Du } } +func (sc *StreamCapture) WaitForData(t testing.TB, substr string, timeout time.Duration) SSEEvent { + t.Helper() + deadline := time.After(timeout) + + for { + select { + case evt := <-sc.events: + if strings.Contains(evt.Data, substr) { + return evt + } + case <-deadline: + t.Fatalf("ListenSSE: timed out after %v waiting for data containing %q", timeout, substr) + return SSEEvent{} + } + } +} + +// Endpoint conveniently extracts an MCP SSE endpoint event. +func (sc *StreamCapture) Endpoint(t testing.TB, timeout time.Duration) string { + t.Helper() + evt := sc.WaitFor(t, "endpoint", timeout) + return strings.TrimSpace(evt.Data) +} + func (sc *StreamCapture) Close() { sc.cancel() } diff --git a/sse_test.go b/sse_test.go index 1515ab4..9b6aef0 100644 --- a/sse_test.go +++ b/sse_test.go @@ -2,6 +2,9 @@ package testkit_test import ( "context" + "fmt" + "net/http" + "net/http/httptest" "strings" "testing" "time" @@ -45,3 +48,79 @@ func TestSSE_StreamCaptureAndWaitFor(t *testing.T) { t.Fatalf("expected SSE payload containing 'order_confirmed', got: %q", evt.Data) } } + +func TestSSE_EndpointHelper(t *testing.T) { + // Minimal MCP-style SSE handler + mux := http.NewServeMux() + mux.HandleFunc("GET /mcp/sse", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.WriteHeader(http.StatusOK) + + flusher, ok := w.(http.Flusher) + if !ok { + t.Error("streaming unsupported") + return + } + + _, _ = fmt.Fprintf(w, "event: endpoint\ndata: /mcp/message?sessionId=abc123\n\n") + flusher.Flush() + + <-r.Context().Done() // keep connection open until cleanup + }) + + suite := testkit.NewWithHandler(t, mux) + stream := suite.ListenSSE("/mcp/sse") + defer stream.Close() + + endpoint := stream.Endpoint(t, 2*time.Second) + + if !strings.Contains(endpoint, "/mcp/message?sessionId=") { + t.Fatalf("unexpected endpoint: %s", endpoint) + } +} + +func TestSSE_ListenSSEWithRequest_Post(t *testing.T) { + t.Parallel() + + mux := http.NewServeMux() + mux.HandleFunc("POST /rpc/stream", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if got := r.Header.Get("Accept"); got != "text/event-stream" { + http.Error(w, "bad accept", http.StatusBadRequest) + return + } + + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + _, _ = fmt.Fprintf(w, "event: message\ndata: {\"status\":\"ok\"}\n\n") + flusher.Flush() + + <-r.Context().Done() + }) + + suite := testkit.NewWithHandler(t, mux) + + req := httptest.NewRequest(http.MethodPost, "/rpc/stream", + strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"ping"}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + + stream := suite.ListenSSEWithRequest(t, req) + defer stream.Close() + + evt := stream.WaitFor(t, "message", 2*time.Second) + + if !strings.Contains(evt.Data, `"status":"ok"`) { + t.Fatalf("unexpected SSE data: %q", evt.Data) + } +}