Coverage for vibefpmgui/dap.py: 84%

1"""A Debug Adapter Protocol client: one debug adapter (gdb's ``--interpreter=dap``, 

2``lldb-dap``…) run as a process, spoken to over its stdin and stdout (see 

3`vibefpmgui.wire` for the framing; `vibefpmgui.debug_session` drives it). 

4 

5Each request gets its answer through a callback, ``callback(success, body, 

6message)``, called once: with the adapter's answer, or, when it doesn't 

7answer in time or stops, with success False and why. Events come as the 

8`event` signal. A request from the adapter (``runInTerminal``…) is answered 

9"not supported". 

10 

11As for fortls (`vibefpmgui.lsp`): no threads (the process is read in the GUI 

12thread), and methods, not lambdas, connected to the process. 

13""" 

14 

15from __future__ import annotations 

16 

17from collections.abc import Callable 

18 

19from qtpy.QtCore import QObject, QProcess, QProcessEnvironment, QTimer, Signal 

20 

21from .wire import FrameReader, frame 

22 

23TIMEOUT = 10_000 # ms a request may take (a hung adapter can't leave a session stuck) 

24 

25Callback = Callable[[bool, dict, str], None] 

26 

27 

28class DapClient(QObject): 

29 event = Signal(str, object) # an event's name and body ({} if none) 

30 stopped = Signal(str) # the adapter's process ended (why), not by stop() 

31 

32 def __init__(self, parent=None): 

33 super().__init__(parent) 

34 self.process: QProcess | None = None 

35 self._reader = FrameReader() 

36 self._seq = 0 

37 self._pending: dict[int, tuple[str, Callback, QTimer]] = {} 

38 self.capabilities: dict = {} 

39 

40 # ---------------------------------------------------------- lifecycle 

41 def is_running(self) -> bool: 

42 return self.process is not None 

43 

44 def start(self, command: list[str], cwd: str, env: dict[str, str] | None = None) -> bool: 

45 """Start the adapter (``command``: its program and arguments).""" 

46 self.stop() 

47 process = QProcess(self) 

48 process.setWorkingDirectory(cwd) 

49 if env: 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 environment = QProcessEnvironment.systemEnvironment() 

51 for key, value in env.items(): 

52 environment.insert(key, value) 

53 process.setProcessEnvironment(environment) 

54 process.setStandardErrorFile(QProcess.nullDevice()) 

55 process.readyReadStandardOutput.connect(self._read) 

56 process.finished.connect(self._process_finished) 

57 process.errorOccurred.connect(self._process_error) 

58 self.process = process 

59 process.start(command[0], command[1:]) 

60 return True 

61 

62 def stop(self): 

63 """End the adapter (and with it the program it debugs), at once.""" 

64 process, self.process = self.process, None 

65 self._fail_pending("the debugger was stopped") 

66 self._reader.clear() 

67 if process is not None: 

68 process.kill() 

69 process.waitForFinished(1000) 

70 process.deleteLater() 

71 

72 def _process_finished(self, *_): 

73 if isinstance(self.sender(), QProcess): 73 ↛ exitline 73 didn't return from function '_process_finished' because the condition on line 73 was always true

74 self._ended(self.sender()) 

75 

76 def _process_error(self, error): 

77 if error == QProcess.FailedToStart and isinstance(self.sender(), QProcess): 

78 self._ended(self.sender()) 

79 

80 def _ended(self, process: QProcess): 

81 if process is not self.process: 

82 return # one we stopped 

83 self.process = None 

84 why = ("the debugger could not be started" if process.error() == QProcess.FailedToStart 

85 else f"the debugger stopped (exit code {process.exitCode()})") 

86 self._fail_pending(why) 

87 try: 

88 process.deleteLater() 

89 except RuntimeError: 

90 return # the app is shutting down 

91 self.stopped.emit(why) 

92 

93 # ---------------------------------------------------------- messages 

94 def request(self, command: str, arguments: dict | None = None, callback: Callback | None = None, 

95 timeout: int = TIMEOUT) -> bool: 

96 """Send a request; False if no adapter is running (``callback`` is 

97 then called at once, with success False).""" 

98 if self.process is None: 

99 if callback is not None: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 callback(False, {}, "no debugger is running") 

101 return False 

102 self._seq += 1 

103 seq = self._seq 

104 timer = QTimer(self) 

105 timer.setSingleShot(True) 

106 timer.setInterval(timeout) 

107 timer.timeout.connect(self._timed_out) 

108 timer.setProperty("seq", seq) 

109 self._pending[seq] = (command, callback or _ignore, timer) 

110 if timeout > 0: 

111 timer.start() 

112 self.process.write(frame({"seq": seq, "type": "request", "command": command, 

113 "arguments": arguments or {}})) 

114 return True 

115 

116 def _timed_out(self): 

117 seq = self.sender().property("seq") if self.sender() is not None else None 

118 entry = self._pending.pop(seq, None) 

119 if entry is not None: 119 ↛ exitline 119 didn't return from function '_timed_out' because the condition on line 119 was always true

120 command, callback, timer = entry 

121 timer.deleteLater() 

122 _call(callback, False, {}, f"the debugger didn't answer {command} in time") 

123 

124 def _fail_pending(self, why: str): 

125 pending, self._pending = self._pending, {} 

126 for _command, callback, timer in pending.values(): 

127 timer.stop() 

128 timer.deleteLater() 

129 _call(callback, False, {}, why) 

130 

131 def _read(self): 

132 if self.process is None: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true

133 return 

134 for message in self._reader.feed(bytes(self.process.readAllStandardOutput())): 

135 self._dispatch(message) 

136 

137 def _dispatch(self, message: dict): 

138 kind = message.get("type") 

139 if kind == "response": 

140 entry = self._pending.pop(message.get("request_seq"), None) 

141 if entry is None: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 return 

143 command, callback, timer = entry 

144 timer.stop() 

145 timer.deleteLater() 

146 if command == "initialize" and message.get("success"): 

147 self.capabilities = message.get("body") or {} 

148 _call(callback, bool(message.get("success")), message.get("body") or {}, 

149 str(message.get("message") or "")) 

150 elif kind == "event": 150 ↛ 152line 150 didn't jump to line 152 because the condition on line 150 was always true

151 self.event.emit(str(message.get("event") or ""), message.get("body") or {}) 

152 elif kind == "request" and self.process is not None: # (runInTerminal…: not supported yet) 

153 self._seq += 1 

154 self.process.write(frame({"seq": self._seq, "type": "response", "request_seq": message.get("seq"), 

155 "command": message.get("command"), "success": False, 

156 "message": "not supported"})) 

157 

158 

159def _ignore(*_): 

160 pass 

161 

162 

163def _call(callback: Callback, success: bool, body: dict, message: str): 

164 try: 

165 callback(success, body, message) 

166 except RuntimeError: 

167 pass # what asked has been deleted (a closed panel)