From 86039811ce1bc95b0aeb36fbfeb5b5131604ac1f Mon Sep 17 00:00:00 2001 From: pablu Date: Mon, 13 Jul 2026 18:33:39 +0200 Subject: [PATCH] wip, rework bridge --- .gitignore | 1 + internal/bridge/client.go | 679 ++++++++++++++++++------------- internal/bridge/client_test.go | 354 ++++++++++++++++ internal/bridge/protocol.go | 173 ++++++-- internal/bridge/protocol_test.go | 214 ++++++++++ python/pybug_runtime.py | 24 +- 6 files changed, 1137 insertions(+), 308 deletions(-) create mode 100644 internal/bridge/client_test.go create mode 100644 internal/bridge/protocol_test.go diff --git a/.gitignore b/.gitignore index a6184bd..e5123d7 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ app.log +.direnv/ diff --git a/internal/bridge/client.go b/internal/bridge/client.go index d3ebaca..9f2c3cd 100644 --- a/internal/bridge/client.go +++ b/internal/bridge/client.go @@ -1,15 +1,15 @@ package bridge import ( - "bufio" + "context" "encoding/json" "errors" "fmt" "io" - "log/slog" + "os" "os/exec" - "slices" "sync" + "sync/atomic" ) var ErrNotRunning = errors.New("bridge not running") @@ -21,349 +21,480 @@ type ExecutionPoint struct { } type Bridge struct { - stdin io.Writer - stdout *bufio.Reader - - cmd *exec.Cmd - - input chan string - - outputLock *sync.RWMutex - output []chan string - - executionStopLock *sync.RWMutex - executionStop []chan ExecutionPoint - - registry map[string]chan string - registryLock *sync.RWMutex - - breakpoints map[string][]int - - callbacksLock *sync.RWMutex - callbacks map[string]map[int]func() - - running bool - path string + + lifecycleMu sync.Mutex + started bool + finished bool + closing bool + failure error + exitErr error + + cmd *exec.Cmd + stdin io.WriteCloser + protocol io.ReadCloser + encoder *json.Encoder + decoder *json.Decoder + + writeMu sync.Mutex + + nextID atomic.Uint64 + pendingMu sync.Mutex + pending map[uint64]chan Message + + events chan Message + output chan Output + done chan struct{} + + finishOnce sync.Once +} + +type OutputStream string + +const ( + StdoutStream OutputStream = "stdout" + StderrStream OutputStream = "stderr" +) + +type Output struct { + Stream OutputStream + Text string } func NewBridge(path string) *Bridge { return &Bridge{ - path: path, - input: make(chan string), - registry: make(map[string]chan string), - registryLock: &sync.RWMutex{}, - breakpoints: make(map[string][]int), - - callbacksLock: &sync.RWMutex{}, - callbacks: make(map[string]map[int]func()), - - output: make([]chan string, 0), - outputLock: &sync.RWMutex{}, - - executionStop: make([]chan ExecutionPoint, 0), - executionStopLock: &sync.RWMutex{}, - - running: false, + path: path, + pending: make(map[uint64]chan Message), + events: make(chan Message, 32), + output: make(chan Output, 128), + done: make(chan struct{}), } } +func (b *Bridge) Events() <-chan Message { + return b.events +} + +func (b *Bridge) Output() <-chan Output { + return b.output +} + +func (b *Bridge) Done() <-chan struct{} { + return b.done +} + func (b *Bridge) Start() error { - if b.running { + b.lifecycleMu.Lock() + defer b.lifecycleMu.Unlock() + + if b.started { return ErrAlreadyStarted } - b.cmd = exec.Command( + protocolReader, protocolWriter, err := os.Pipe() + if err != nil { + return fmt.Errorf("create protocol pipe: %w", err) + } + + cmd := exec.Command( "python", "-u", "python/pybug_runtime.py", b.path, ) - var err error - b.stdin, err = b.cmd.StdinPipe() + cmd.ExtraFiles = []*os.File{protocolWriter} + cmd.Env = append(os.Environ(), "PYBUG_PROTOCOL_FD=3") + + stdin, err := cmd.StdinPipe() if err != nil { - return err + + protocolWriter.Close() + protocolReader.Close() + return fmt.Errorf("create command pipe: %w", err) } - reader, err := b.cmd.StdoutPipe() + stdout, err := cmd.StdoutPipe() if err != nil { - return err - } - b.stdout = bufio.NewReader(reader) + stdin.Close() - err = b.cmd.Start() - if err != nil { - return err + protocolReader.Close() + protocolWriter.Close() + return fmt.Errorf("create stdout pipe: %w", err) } - b.running = true - go b.readLoop() - go b.writeLoop() + stderr, err := cmd.StderrPipe() + if err != nil { + stdout.Close() + protocolReader.Close() + protocolWriter.Close() + return fmt.Errorf("create stderr pipe: %w", err) + } + + if err := cmd.Start(); err != nil { + stdin.Close() + stdout.Close() + stderr.Close() + protocolWriter.Close() + protocolReader.Close() + return fmt.Errorf("start debugger runtime: %w", err) + } + + if err := protocolWriter.Close(); err != nil { + cmd.Process.Kill() + stdin.Close() + stdout.Close() + stderr.Close() + protocolWriter.Close() + protocolReader.Close() + return fmt.Errorf("close parent protocol writer: %w", err) + } + + b.cmd = cmd + b.stdin = stdin + b.protocol = protocolReader + b.encoder = json.NewEncoder(stdin) + b.decoder = json.NewDecoder(protocolReader) + b.started = true + + go b.readProtocol() + go b.readOutput(stdout, StdoutStream) + go b.readOutput(stderr, StderrStream) + go b.waitForProcess() return nil } -func (b *Bridge) Subscribe() chan string { - b.outputLock.Lock() - defer b.outputLock.Unlock() +func (b *Bridge) ensureRunning() error { + b.lifecycleMu.Lock() + defer b.lifecycleMu.Unlock() - c := make(chan string) - b.output = append(b.output, c) - - return c -} - -func (b *Bridge) SubscribeStopped() chan ExecutionPoint { - b.executionStopLock.Lock() - defer b.executionStopLock.Unlock() - - c := make(chan ExecutionPoint) - b.executionStop = append(b.executionStop, c) - - return c -} - -func (b *Bridge) Locals() (map[string]any, error) { - if !b.running { - return nil, ErrNotRunning - } - - requestId, cmd := makeCommand(LocalsCommand, map[string]any{}) - - c := b.sendCommand(requestId, cmd) - - obj := <-c - - var m map[string]any - err := json.Unmarshal([]byte(obj), &m) - if err != nil { - return nil, err - } - - vars, ok := m["vars"].(map[string]any) - if !ok { - return nil, errors.New("could not extract vars from response") - } - - return vars, nil -} - -func (b *Bridge) Step() error { - if !b.running { + if !b.started || b.finished { return ErrNotRunning } - _, cmd := makeCommand(StepCommand, nil) - b.sendCommandNoResponse(cmd) + return nil +} + +func (b *Bridge) writeMessage(msg Message) error { + b.writeMu.Lock() + defer b.writeMu.Unlock() + + if err := b.encoder.Encode(msg); err != nil { + return fmt.Errorf("write protocol message: %w", err) + } return nil } -func (b *Bridge) Breakpoint(file string, line int) (set bool, err error) { - if !b.running { - return false, ErrNotRunning +func (b *Bridge) request( + ctx context.Context, + name string, + body any, +) (Message, error) { + if err := b.ensureRunning(); err != nil { + return Message{}, err } - var command CommandType - if _, ok := b.breakpoints[file]; ok && slices.Contains(b.breakpoints[file], line) { - command = UnbreakCommand - } else { - command = BreakCommand + id := b.nextID.Add(1) + response := make(chan Message, 1) + + b.pendingMu.Lock() + b.pending[id] = response + b.pendingMu.Unlock() + + defer func() { + b.pendingMu.Lock() + delete(b.pending, id) + b.pendingMu.Unlock() + }() + + msg, err := newMessage(Request, &id, name, body) + if err != nil { + return Message{}, fmt.Errorf("create %q request: %w", name, err) } - requestId, cmd := makeCommand(command, map[string]any{ - "file": file, - "line": line, + if err := msg.Validate(); err != nil { + return Message{}, fmt.Errorf("validate %q request: %w", name, err) + } + + if err := b.writeMessage(msg); err != nil { + return Message{}, err + } + + select { + case response := <-response: + if response.Name != name { + return Message{}, fmt.Errorf("response name mismatch: requested %q, received %q", name, response.Name) + } + + if response.Error != nil { + return Message{}, response.Error + } + return response, nil + case <-ctx.Done(): + return Message{}, ctx.Err() + case <-b.done: + return Message{}, ErrNotRunning + } +} + +func (b *Bridge) deliverResponse(msg Message) { + id := *msg.RequestId + + b.pendingMu.Lock() + response, ok := b.pending[id] + b.pendingMu.Unlock() + + if !ok { + return + } + + select { + case response <- msg: + case <-b.done: + default: + } +} + +func (b *Bridge) readProtocol() { + for { + var msg Message + if err := b.decoder.Decode(&msg); err != nil { + if errors.Is(err, io.EOF) { + return + } + + select { + case <-b.done: + return + default: + } + + b.abort(fmt.Errorf("decode protocol message: %w", err)) + return + } + + if err := msg.Validate(); err != nil { + b.abort(fmt.Errorf("invalid protocol message: %w", err)) + return + } + + switch msg.Type { + case Response: + b.deliverResponse(msg) + case Event: + select { + case b.events <- msg: + case <-b.done: + return + } + default: + b.abort(fmt.Errorf("unexpected message type from runtime: %q", msg.Type)) + return + } + } +} + +func (b *Bridge) readOutput(reader io.ReadCloser, stream OutputStream) { + defer reader.Close() + + buffer := make([]byte, 4096) + + for { + n, err := reader.Read(buffer) + + if n > 0 { + output := Output{ + Stream: stream, + Text: string(buffer[:n]), + } + + select { + case b.output <- output: + case <-b.done: + return + } + } + + if err != nil { + if !errors.Is(err, io.EOF) { + b.abort(fmt.Errorf("read target %s: %w", stream, err)) + } + return + } + } +} + +func (b *Bridge) waitForProcess() { + err := b.cmd.Wait() + b.finish(err) +} + +func (b *Bridge) abort(err error) { + b.lifecycleMu.Lock() + if b.failure == nil { + b.failure = err + } + + var process *os.Process + if b.cmd != nil { + process = b.cmd.Process + } + finished := b.finished + + b.lifecycleMu.Unlock() + + if !finished && process != nil { + _ = process.Kill() + } +} + +func (b *Bridge) finish(waitErr error) { + b.finishOnce.Do(func() { + b.lifecycleMu.Lock() + + b.finished = true + if b.failure != nil { + b.exitErr = b.failure + } else if !b.closing { + b.exitErr = waitErr + } + + stdin := b.stdin + protocol := b.protocol + + b.lifecycleMu.Unlock() + + if stdin != nil { + _ = stdin.Close() + } + + if protocol != nil { + _ = protocol.Close() + } + + close(b.done) }) - - c := b.sendCommand(requestId, cmd) - - obj := <-c - - var m map[string]any - err = json.Unmarshal([]byte(obj), &m) - if err != nil { - return false, err - } - - if m["status"] != "ok" { - return false, fmt.Errorf("error occured on break, err: %s", m["error"]) - } - - if command == BreakCommand { - b.breakpoints[file] = append(b.breakpoints[file], line) - return true, nil - } else { - breakpointsLen := len(b.breakpoints[file]) - index := slices.Index(b.breakpoints[file], line) - b.breakpoints[file][index] = b.breakpoints[file][breakpointsLen-1] - - b.breakpoints[file] = b.breakpoints[file][0 : breakpointsLen-1] - return false, nil - } } -func (b *Bridge) Continue() error { - if !b.running { +func (b *Bridge) Wait(ctx context.Context) error { + b.lifecycleMu.Lock() + started := b.started + b.lifecycleMu.Unlock() + + if !started { return ErrNotRunning } - _, cmd := makeCommand(ContinueCommand, map[string]any{}) + select { + case <-b.done: + return b.processError() - b.sendCommandNoResponse(cmd) - return nil -} - -func (b *Bridge) sendCommandNoResponse(command string) { - b.input <- command -} - -func (b *Bridge) sendCommand(requestId string, command string) chan string { - b.registryLock.Lock() - defer b.registryLock.Unlock() - - channel := make(chan string) - - b.registry[requestId] = channel - b.input <- command - - return channel -} - -func (b *Bridge) writeLoop() { - slog.Debug("started writeLoop") - defer slog.Debug("writeLoop exited") - for { - cmd := <-b.input - - if !b.running { - return - } - - slog.Info("Received command", "cmd", cmd) - - _, err := b.stdin.Write([]byte(cmd + "\n")) - if err != nil { - slog.Error("Error occured while writing to stdin", "error", err) - } - - slog.Debug("Command written", "cmd", cmd) + case <-ctx.Done(): + return ctx.Err() } } -func (b *Bridge) readLoop() { - slog.Debug("started readLoop") - defer slog.Debug("readLoop exited") - for { - slog.Debug("reading string from stdout waiting for newline") - line, err := b.stdout.ReadString('\n') - if err != nil && !errors.Is(err, io.EOF) { - slog.Error("Error occured while reading from stdout", "error", err) - continue - } else if errors.Is(err, io.EOF) { - b.running = false - return - } +func (b *Bridge) Close(ctx context.Context) error { + b.lifecycleMu.Lock() - var msg map[string]any - err = json.Unmarshal([]byte(line), &msg) - if err != nil { - slog.Debug("read line from stdout", "line", line) + if !b.started { + b.lifecycleMu.Unlock() + return ErrNotRunning + } - b.outputLock.RLock() - for _, c := range b.output { - c <- line - } - b.outputLock.RUnlock() + if b.finished { + b.lifecycleMu.Unlock() + return b.processError() + } - } else if requestId, ok := msg["request_id"].(string); ok { - b.registryLock.RLock() - c, ok := b.registry[requestId] - b.registryLock.RUnlock() + b.closing = true + process := b.cmd.Process + b.lifecycleMu.Unlock() - if !ok { - slog.Error("Could not find requestId in registry", "requestId", requestId) - continue - } - - c <- line - } else { - b.handleStopped(msg) + if process != nil { + err := process.Kill() + if err != nil && !errors.Is(err, os.ErrProcessDone) { + return fmt.Errorf("kill debugger runtime: %w", err) } } + + return b.Wait(ctx) } -func (b *Bridge) handleStopped(msg map[string]any) error { - // TODO: set to stopped - if event, ok := msg["event"]; !ok || event != "stopped" { - slog.Warn("received unkown event", "msg", msg) - return errors.New("unknown event encountered") +func (b *Bridge) processError() error { + b.lifecycleMu.Lock() + defer b.lifecycleMu.Unlock() + + return b.exitErr +} + +func (b *Bridge) Locals(ctx context.Context) (LocalsResponse, error) { + msg, err := b.request(ctx, "locals/get", nil) + if err != nil { + return LocalsResponse{}, err } - file := msg["file"].(string) - line, ok := toInt(msg["line"]) - if !ok { - slog.Error("could not convert line to int", "line", msg["line"]) - return errors.New("could not convert line to int") + var result LocalsResponse + if err := json.Unmarshal(msg.Body, &result); err != nil { + return LocalsResponse{}, fmt.Errorf("decode locals response: %w", err) } - slog.Info("received stopped event") - b.callbacksLock.RLock() - defer b.callbacksLock.RUnlock() + return result, nil +} - if callback, ok := b.callbacks[file][line]; ok { - slog.Info("found callback, now running", "file", file, "line", line) - go callback() - } +func (b *Bridge) Continue(ctx context.Context) error { + _, err := b.request(ctx, "execution/continue", nil) + return err +} - b.executionStopLock.RLock() - defer b.executionStopLock.RUnlock() +func (b *Bridge) StepInto(ctx context.Context) error { + _, err := b.request(ctx, "execution/step-into", nil) + return err +} - for _, c := range b.executionStop { - c <- ExecutionPoint{ +func (b *Bridge) StepOver(ctx context.Context) error { + _, err := b.request(ctx, "execution/step-over", nil) + return err +} + +func (b *Bridge) SetBreakpoint(ctx context.Context, file string, line int) (SetBreakpointResponse, error) { + msg, err := b.request( + ctx, + "breakpoint/set", + SetBreakpointRequest{ File: file, Line: line, - } + }, + ) + if err != nil { + return SetBreakpointResponse{}, err } - return nil -} - -func (b *Bridge) OnBreakpoint(file string, line int, callback func()) { - b.callbacksLock.Lock() - defer b.callbacksLock.Unlock() - - if f, ok := b.callbacks[file]; ok { - f[line] = callback - } else { - b.callbacks[file] = map[int]func(){ - line: callback, - } + var response SetBreakpointResponse + if err := json.Unmarshal(msg.Body, &response); err != nil { + return SetBreakpointResponse{}, fmt.Errorf( + "decode breakpoint response: %w", err, + ) } + + return response, nil } -func (b *Bridge) Wait() error { - return b.cmd.Wait() -} +func (b *Bridge) RemoveBreakpoint( + ctx context.Context, + file string, + line int, +) error { + _, err := b.request( + ctx, + "breakpoint/remove", + RemoveBreakpointRequest{ + File: file, + Line: line, + }, + ) -func toInt(v any) (int, bool) { - switch x := v.(type) { - case int: - return x, true - case int8: - return int(x), true - case int16: - return int(x), true - case int32: - return int(x), true - case int64: - return int(x), true - case float32: - return int(x), true - case float64: - return int(x), true - default: - return 0, false - } + return err } diff --git a/internal/bridge/client_test.go b/internal/bridge/client_test.go new file mode 100644 index 0000000..9fdd21b --- /dev/null +++ b/internal/bridge/client_test.go @@ -0,0 +1,354 @@ +package bridge + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +type bridgeHarness struct { + bridge *Bridge + requestDecoder *json.Decoder + responseEncoder *json.Encoder + requestReader *io.PipeReader + responseWriter *io.PipeWriter +} + +func newBridgeHarness(t *testing.T) *bridgeHarness { + t.Helper() + + requestReader, requestWriter := io.Pipe() + responseReader, responseWriter := io.Pipe() + + b := NewBridge("unused.py") + b.started = true + b.stdin = requestWriter + b.protocol = responseReader + b.encoder = json.NewEncoder(requestWriter) + b.decoder = json.NewDecoder(responseReader) + + h := &bridgeHarness{ + bridge: b, + requestDecoder: json.NewDecoder(requestReader), + responseEncoder: json.NewEncoder(responseWriter), + requestReader: requestReader, + responseWriter: responseWriter, + } + + go b.readProtocol() + + t.Cleanup(func() { + responseWriter.Close() + requestReader.Close() + b.finish(nil) + }) + + return h +} + +func testContext(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + t.Cleanup(cancel) + return ctx +} + +func (h *bridgeHarness) receiveRequest(t *testing.T) Message { + t.Helper() + + var request Message + if err := h.requestDecoder.Decode(&request); err != nil { + t.Fatalf("decode request: %v", err) + } + return request +} + +func (h *bridgeHarness) sendMessage(t *testing.T, msg Message) { + t.Helper() + + if err := h.responseEncoder.Encode(msg); err != nil { + t.Fatalf("encode runtime message: %v", err) + } +} + +func TestRequestCorrelatesResponse(t *testing.T) { + h := newBridgeHarness(t) + + type requestResult struct { + message Message + err error + } + resultChannel := make(chan requestResult, 1) + ctx := testContext(t) + + go func() { + msg, err := h.bridge.request(ctx, "locals/get", nil) + resultChannel <- requestResult{message: msg, err: err} + }() + + request := h.receiveRequest(t) + if request.Type != Request || request.Name != "locals/get" || request.RequestId == nil { + t.Fatalf("unexpected request: %+v", request) + } + + response, err := newMessage(Response, request.RequestId, request.Name, LocalsResponse{ + Variables: map[string]string{"answer": "42"}, + }) + if err != nil { + t.Fatalf("newMessage() error = %v", err) + } + h.sendMessage(t, response) + + result := <-resultChannel + if result.err != nil { + t.Fatalf("request() error = %v", result.err) + } + + var body LocalsResponse + if err := json.Unmarshal(result.message.Body, &body); err != nil { + t.Fatalf("decode response body: %v", err) + } + if body.Variables["answer"] != "42" { + t.Fatalf("response variables = %#v", body.Variables) + } +} + +func TestConcurrentRequestsAreCorrelatedByID(t *testing.T) { + h := newBridgeHarness(t) + ctx := testContext(t) + + type requestResult struct { + name string + body string + err error + } + + const requestCount = 8 + results := make(chan requestResult, requestCount) + + for i := range requestCount { + name := fmt.Sprintf("test/request-%d", i) + go func() { + msg, err := h.bridge.request(ctx, name, nil) + if err != nil { + results <- requestResult{name: name, err: err} + return + } + + var body struct { + Name string `json:"name"` + } + err = json.Unmarshal(msg.Body, &body) + results <- requestResult{name: name, body: body.Name, err: err} + }() + } + + requests := make([]Message, 0, requestCount) + ids := make(map[uint64]struct{}, requestCount) + for range requestCount { + request := h.receiveRequest(t) + if request.RequestId == nil { + t.Fatalf("request has no ID: %+v", request) + } + if _, exists := ids[*request.RequestId]; exists { + t.Fatalf("duplicate request ID %d", *request.RequestId) + } + ids[*request.RequestId] = struct{}{} + requests = append(requests, request) + } + + for i := len(requests) - 1; i >= 0; i-- { + request := requests[i] + response, err := newMessage(Response, request.RequestId, request.Name, struct { + Name string `json:"name"` + }{Name: request.Name}) + if err != nil { + t.Fatalf("newMessage() error = %v", err) + } + h.sendMessage(t, response) + } + + for range requestCount { + result := <-results + if result.err != nil { + t.Fatalf("request %q error = %v", result.name, result.err) + } + if result.body != result.name { + t.Fatalf("request %q received body for %q", result.name, result.body) + } + } +} + +func TestRequestReturnsProtocolError(t *testing.T) { + h := newBridgeHarness(t) + errChannel := make(chan error, 1) + ctx := testContext(t) + + go func() { + _, err := h.bridge.request(ctx, "execution/continue", nil) + errChannel <- err + }() + + request := h.receiveRequest(t) + h.sendMessage(t, newErrorResponse( + *request.RequestId, + request.Name, + "not_paused", + "execution is not paused", + )) + + err := <-errChannel + var protocolErr *ProtocolError + if !errors.As(err, &protocolErr) { + t.Fatalf("request() error = %v, want ProtocolError", err) + } + if protocolErr.Code != "not_paused" { + t.Fatalf("protocol error code = %q", protocolErr.Code) + } +} + +func TestRequestRejectsMismatchedResponseName(t *testing.T) { + h := newBridgeHarness(t) + errChannel := make(chan error, 1) + ctx := testContext(t) + + go func() { + _, err := h.bridge.request(ctx, "locals/get", nil) + errChannel <- err + }() + + request := h.receiveRequest(t) + response, err := newMessage(Response, request.RequestId, "execution/continue", nil) + if err != nil { + t.Fatalf("newMessage() error = %v", err) + } + h.sendMessage(t, response) + + err = <-errChannel + if err == nil || !strings.Contains(err.Error(), "response name mismatch") { + t.Fatalf("request() error = %v, want response name mismatch", err) + } +} + +func TestCancelledRequestIsRemovedFromRegistry(t *testing.T) { + h := newBridgeHarness(t) + ctx, cancel := context.WithCancel(context.Background()) + errChannel := make(chan error, 1) + + go func() { + _, err := h.bridge.request(ctx, "locals/get", nil) + errChannel <- err + }() + + request := h.receiveRequest(t) + cancel() + + if err := <-errChannel; !errors.Is(err, context.Canceled) { + t.Fatalf("request() error = %v, want context.Canceled", err) + } + + h.bridge.pendingMu.Lock() + _, exists := h.bridge.pending[*request.RequestId] + h.bridge.pendingMu.Unlock() + if exists { + t.Fatalf("request ID %d remained in pending registry", *request.RequestId) + } +} + +func TestReadProtocolPublishesEvent(t *testing.T) { + h := newBridgeHarness(t) + event, err := newMessage(Event, nil, "execution/stopped", ExecutionStoppedEvent{ + Reason: "step", + File: "test.py", + Line: 12, + }) + if err != nil { + t.Fatalf("newMessage() error = %v", err) + } + h.sendMessage(t, event) + + select { + case received := <-h.bridge.Events(): + if received.Name != "execution/stopped" { + t.Fatalf("event name = %q", received.Name) + } + + var body ExecutionStoppedEvent + if err := json.Unmarshal(received.Body, &body); err != nil { + t.Fatalf("decode event body: %v", err) + } + if body.Reason != "step" || body.File != "test.py" || body.Line != 12 { + t.Fatalf("event body = %+v", body) + } + + case <-time.After(time.Second): + t.Fatal("timed out waiting for event") + } +} + +func TestReadOutputPublishesStreamAndText(t *testing.T) { + b := NewBridge("unused.py") + reader := io.NopCloser(strings.NewReader("hello from target\n")) + + go b.readOutput(reader, StdoutStream) + + select { + case output := <-b.Output(): + if output.Stream != StdoutStream { + t.Fatalf("output stream = %q", output.Stream) + } + if output.Text != "hello from target\n" { + t.Fatalf("output text = %q", output.Text) + } + + case <-time.After(time.Second): + t.Fatal("timed out waiting for output") + } +} + +func TestRequestBeforeStartReturnsErrNotRunning(t *testing.T) { + b := NewBridge("unused.py") + + _, err := b.request(testContext(t), "locals/get", nil) + if !errors.Is(err, ErrNotRunning) { + t.Fatalf("request() error = %v, want ErrNotRunning", err) + } +} + +func TestCloseTerminatesRunningProcess(t *testing.T) { + if os.Getenv("PYBUG_BRIDGE_HELPER_PROCESS") == "1" { + time.Sleep(time.Minute) + return + } + + cmd := exec.Command(os.Args[0], "-test.run=TestCloseTerminatesRunningProcess") + cmd.Env = append(os.Environ(), "PYBUG_BRIDGE_HELPER_PROCESS=1") + if err := cmd.Start(); err != nil { + t.Fatalf("start helper process: %v", err) + } + + b := NewBridge("unused.py") + b.cmd = cmd + b.started = true + go b.waitForProcess() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + if err := b.Close(ctx); err != nil { + t.Fatalf("Close() error = %v", err) + } + + select { + case <-b.Done(): + default: + t.Fatal("Done channel was not closed") + } +} diff --git a/internal/bridge/protocol.go b/internal/bridge/protocol.go index f8ddb6b..aacaa4b 100644 --- a/internal/bridge/protocol.go +++ b/internal/bridge/protocol.go @@ -2,43 +2,156 @@ package bridge import ( "encoding/json" - "maps" - "math/rand" + "errors" + "fmt" ) -type CommandType string +const ProtocolVersion = 1 + +type MessageType string const ( - ContinueCommand CommandType = "continue" - BreakCommand CommandType = "break" - UnbreakCommand CommandType = "unbreak" - LocalsCommand CommandType = "locals" - StepCommand CommandType = "step" + Request MessageType = "request" + Response MessageType = "response" + Event MessageType = "event" ) -func makeCommand(cmd CommandType, values map[string]any) (requestId string, request string) { - m := make(map[string]any, len(values)+2) - maps.Copy(m, values) - - requestId = randString(8) - - m["cmd"] = cmd - m["request_id"] = requestId - - b, err := json.Marshal(m) - if err != nil { - panic("failed to marshal command " + err.Error()) - } - - return requestId, string(b) +type Message struct { + Version int `json:"version"` + Type MessageType `json:"type"` + RequestId *uint64 `json:"id,omitempty"` + Name string `json:"name"` + Body json.RawMessage `json:"body,omitempty"` + Error *ProtocolError `json:"error,omitempty"` } -const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" - -func randString(n int) string { - b := make([]byte, n) - for i := range b { - b[i] = letterBytes[rand.Intn(len(letterBytes))] +func (m Message) Validate() error { + if m.Version != ProtocolVersion { + return fmt.Errorf("unsupported protocol version %d", m.Version) + } + + if m.Name == "" { + return errors.New("message name is required") + } + + switch m.Type { + case Request: + if m.RequestId == nil { + return errors.New("request ID is required") + } + if m.Error != nil { + return errors.New("request cannot contain an error") + } + case Response: + if m.RequestId == nil { + return errors.New("response ID is required") + } + if m.Error != nil && len(m.Body) != 0 { + return errors.New("response cannot contain both body and error") + } + + case Event: + if m.RequestId != nil { + return errors.New("event cannot contain a request ID") + } + if m.Error != nil { + return errors.New("event cannot contain an error") + } + default: + return fmt.Errorf("unknown message type %q", m.Type) + } + + return nil +} + +type ProtocolError struct { + Code string `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` +} + +func (e *ProtocolError) Error() string { + if e == nil { + return "" + } + + if e.Code == "" { + return e.Message + } + + return fmt.Sprintf("%s: %s", e.Code, e.Message) +} + +type SetBreakpointRequest struct { + File string `json:"file"` + Line int `json:"line"` +} + +type SetBreakpointResponse struct { + File string `json:"file"` + Line int `json:"line"` + Verified bool `json:"verified"` +} + +type RemoveBreakpointRequest struct { + File string `json:"file"` + Line int `json:"line"` +} + +type LocalsResponse struct { + Variables map[string]string `json:"variables"` +} + +type ExecutionStoppedEvent struct { + Reason string `json:"reason"` + File string `json:"file"` + Line int `json:"line"` +} + +type ExecutionExitedEvent struct { + ExitCode int `json:"exit_code"` +} + +func newMessage( + messageType MessageType, + id *uint64, + name string, + body any, +) (Message, error) { + var raw json.RawMessage + + if body != nil { + encoded, err := json.Marshal(body) + if err != nil { + return Message{}, err + } + + raw = encoded + } + + return Message{ + Version: ProtocolVersion, + Type: messageType, + RequestId: id, + Name: name, + Body: raw, + }, nil +} + +func newErrorResponse( + id uint64, + name string, + code string, + message string, +) Message { + return Message{ + Version: ProtocolVersion, + Type: Response, + RequestId: &id, + Name: name, + Error: &ProtocolError{ + Code: code, + Message: message, + }, } - return string(b) } diff --git a/internal/bridge/protocol_test.go b/internal/bridge/protocol_test.go new file mode 100644 index 0000000..85e77bf --- /dev/null +++ b/internal/bridge/protocol_test.go @@ -0,0 +1,214 @@ +package bridge + +import ( + "encoding/json" + "strings" + "testing" +) + +func messageID(id uint64) *uint64 { + return &id +} + +func TestMessageValidate(t *testing.T) { + tests := []struct { + name string + message Message + wantErr string + }{ + { + name: "request", + message: Message{ + Version: ProtocolVersion, + Type: Request, + RequestId: messageID(1), + Name: "locals/get", + }, + }, + { + name: "successful response", + message: Message{ + Version: ProtocolVersion, + Type: Response, + RequestId: messageID(1), + Name: "locals/get", + Body: json.RawMessage(`{"variables":{}}`), + }, + }, + { + name: "error response", + message: Message{ + Version: ProtocolVersion, + Type: Response, + RequestId: messageID(1), + Name: "locals/get", + Error: &ProtocolError{ + Code: "not_paused", + Message: "execution is not paused", + }, + }, + }, + { + name: "event", + message: Message{ + Version: ProtocolVersion, + Type: Event, + Name: "execution/stopped", + Body: json.RawMessage(`{"reason":"step","file":"test.py","line":1}`), + }, + }, + { + name: "unsupported version", + message: Message{ + Version: ProtocolVersion + 1, + Type: Request, + RequestId: messageID(1), + Name: "locals/get", + }, + wantErr: "unsupported protocol version", + }, + { + name: "missing name", + message: Message{ + Version: ProtocolVersion, + Type: Request, + RequestId: messageID(1), + }, + wantErr: "message name is required", + }, + { + name: "request without ID", + message: Message{ + Version: ProtocolVersion, + Type: Request, + Name: "locals/get", + }, + wantErr: "request ID is required", + }, + { + name: "request with error", + message: Message{ + Version: ProtocolVersion, + Type: Request, + RequestId: messageID(1), + Name: "locals/get", + Error: &ProtocolError{Code: "invalid"}, + }, + wantErr: "request cannot contain an error", + }, + { + name: "response without ID", + message: Message{ + Version: ProtocolVersion, + Type: Response, + Name: "locals/get", + }, + wantErr: "response ID is required", + }, + { + name: "response with body and error", + message: Message{ + Version: ProtocolVersion, + Type: Response, + RequestId: messageID(1), + Name: "locals/get", + Body: json.RawMessage(`{}`), + Error: &ProtocolError{Code: "invalid"}, + }, + wantErr: "response cannot contain both body and error", + }, + { + name: "event with ID", + message: Message{ + Version: ProtocolVersion, + Type: Event, + RequestId: messageID(1), + Name: "execution/stopped", + }, + wantErr: "event cannot contain a request ID", + }, + { + name: "event with error", + message: Message{ + Version: ProtocolVersion, + Type: Event, + Name: "execution/stopped", + Error: &ProtocolError{Code: "invalid"}, + }, + wantErr: "event cannot contain an error", + }, + { + name: "unknown type", + message: Message{ + Version: ProtocolVersion, + Type: MessageType("notification"), + RequestId: messageID(1), + Name: "locals/get", + }, + wantErr: "unknown message type", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.message.Validate() + if tt.wantErr == "" { + if err != nil { + t.Fatalf("Validate() error = %v", err) + } + return + } + + if err == nil { + t.Fatalf("Validate() error = nil, want error containing %q", tt.wantErr) + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Validate() error = %q, want error containing %q", err, tt.wantErr) + } + }) + } +} + +func TestNewMessageEncodesTypedBody(t *testing.T) { + id := uint64(7) + msg, err := newMessage(Request, &id, "breakpoint/set", SetBreakpointRequest{ + File: "example.py", + Line: 42, + }) + if err != nil { + t.Fatalf("newMessage() error = %v", err) + } + + if err := msg.Validate(); err != nil { + t.Fatalf("Validate() error = %v", err) + } + + var body SetBreakpointRequest + if err := json.Unmarshal(msg.Body, &body); err != nil { + t.Fatalf("unmarshal body: %v", err) + } + + if body.File != "example.py" || body.Line != 42 { + t.Fatalf("decoded body = %+v", body) + } +} + +func TestProtocolErrorString(t *testing.T) { + tests := []struct { + name string + err *ProtocolError + want string + }{ + {name: "code and message", err: &ProtocolError{Code: "not_paused", Message: "execution is not paused"}, want: "not_paused: execution is not paused"}, + {name: "message only", err: &ProtocolError{Message: "execution is not paused"}, want: "execution is not paused"}, + {name: "nil", err: nil, want: ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.err.Error(); got != tt.want { + t.Fatalf("Error() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/python/pybug_runtime.py b/python/pybug_runtime.py index 49441a7..9567db0 100644 --- a/python/pybug_runtime.py +++ b/python/pybug_runtime.py @@ -1,23 +1,39 @@ import bdb import json +import os import sys from types import FrameType +PROTOCOL_FD = int(os.environ["PYBUG_PROTOCOL_FD"]) + class PyBugBridgeDebugger(bdb.Bdb): def __init__(self): super().__init__() + self.protocol = os.fdopen( + PROTOCOL_FD, mode="w", encoding="utf-8", buffering=1, closefd=True + ) self.waiting = False def send(self, msg: dict): - sys.stdout.write(json.dumps(msg) + "\n") - sys.stdout.flush() + json.dump(msg, self.protocol, separators=(",", ":")) + self.protocol.write("\n") + self.protocol.flush() def recv(self) -> dict: - return json.loads(sys.stdin.readline()) + line = sys.stdin.readline() + if line == "": + raise EOFError("debugger command channel closed") + return json.loads(line) def user_line(self, frame: FrameType): - print("TRACE:", frame.f_code.co_filename, frame.f_lineno) + print( + "TRACE:", + frame.f_code.co_filename, + frame.f_lineno, + file=sys.stderr, + flush=True, + ) self.send( { "event": "stopped",