eric: comparison DebugClients/Python3/DebugClientBase.py

-:fb9168c2e069
+:8c48f5e0cd92
 import socket
 import select
 import codeop
 import traceback
 import os
+import json
 import imp
 import re
 import atexit
 import signal
-import DebugProtocol
 import DebugClientCapabilities
+import DebugVariables
 from DebugBase import setRecursionLimit, printerr   # __IGNORE_WARNING__
 from AsyncFile import AsyncFile, AsyncPendingWrite
 from DebugConfig import ConfigVarTypeStrings
 from FlexCompleter import Completer
-from DebugUtilities import getargvalues, formatargvalues
+from DebugUtilities import getargvalues, formatargvalues, prepareJsonCommand
 from BreakpointWatch import Breakpoint, Watch
 DebugClientInstance = None
 class DebugClientBase(object):
 """
 Class implementing the client side of the debugger.
 It provides access to the Python interpeter from a debugger running in
-another process whether or not the Qt event loop is running.
+another process.
-The protocol between the debugger and the client assumes that there will be
+The protocol between the debugger and the client is based on JSONRPC 2.0
-a single source of debugger commands and a single source of Python
+PDUs. Each one is sent on a single line, i.e. commands or responses are
-statements.  Commands and statement are always exactly one line and may be
+separated by a linefeed character.
-interspersed.
-The protocol is as follows.  First the client opens a connection to the
-debugger and then sends a series of one line commands.  A command is either
-&gt;Load&lt;, &gt;Step&lt;, &gt;StepInto&lt;, ... or a Python statement.
-See DebugProtocol.py for a listing of valid protocol tokens.
-A Python statement consists of the statement to execute, followed (in a
-separate line) by &gt;OK?&lt;. If the statement was incomplete then the
-response is &gt;Continue&lt;. If there was an exception then the response
-is &gt;Exception&lt;. Otherwise the response is &gt;OK&lt;.  The reason
-for the &gt;OK?&lt; part is to provide a sentinal (ie. the responding
-&gt;OK&lt;) after any possible output as a result of executing the command.
-The client may send any other lines at any other time which should be
-interpreted as program output.
 If the debugger closes the session there is no response from the client.
 The client may close the session at any time as a result of the script
 being debugged closing or crashing.
 <b>Note</b>: This class is meant to be subclassed by individual
 DebugClient classes. Do not instantiate it directly.
 """
 clientCapabilities = DebugClientCapabilities.HasAll
+# keep these in sync with VariablesViewer.VariableItem.Indicators
+Indicators = ("()", "[]", "{:}", "{}")         # __IGNORE_WARNING__
 def __init__(self):
 """
 Constructor
 """
 self.breakpoints = {}
 self.redirect = True
+self.__receiveBuffer = ""
 # The next couple of members are needed for the threaded version.
 # For this base class they contain static values for the non threaded
 # debugger
 # The list of regexp objects to filter variables against
 self.globalsFilterObjects = []
 self.localsFilterObjects = []
-self.pendingResponse = DebugProtocol.ResponseOK
 self._fncache = {}
 self.dircache = []
-self.inRawMode = False
-self.mainProcStr = None     # used for the passive mode
 self.passive = False        # used to indicate the passive mode
 self.running = None
 self.test = None
 self.debugging = False
 def attachThread(self, target=None, args=None, kwargs=None,
 mainThread=False):
 """
 Public method to setup a thread for DebugClient to debug.
-If mainThread is non-zero, then we are attaching to the already
+If mainThread is True, then we are attaching to the already
 started mainthread of the app and the rest of the args are ignored.
 @param target the start function of the target thread (i.e. the user
 code)
 @param args arguments to pass to target
 d["broken"] = self.isBroken()
 else:
 d["broken"] = False
 threadList.append(d)
-self.write("{0}{1!r}\n".format(DebugProtocol.ResponseThreadList,
+self.sendJsonCommand("ResponseThreadList", {
-(currentId, threadList)))
+"currentID": currentId,
+"threadList": threadList,
+})
 def input(self, prompt, echo=True):
 """
 Public method to implement input() using the event loop.
 @param prompt the prompt to be shown (string)
 @param echo Flag indicating echoing of the input (boolean)
 @return the entered string
 """
-self.write("{0}{1!r}\n".format(
+self.sendJsonCommand("RequestRaw", {
-DebugProtocol.ResponseRaw, (prompt, echo)))
+"prompt": prompt,
-self.inRawMode = True
+"echo": echo,
+})
 self.eventLoop(True)
 return self.rawLine
-def __exceptionRaised(self):
-"""
-Private method called in the case of an exception.
-It ensures that the debug server is informed of the raised exception.
-"""
-self.pendingResponse = DebugProtocol.ResponseException
 def sessionClose(self, exit=True):
 """
 Public method to close the session with the debugger and optionally
 terminate.
 try:
 self.set_quit()
 except Exception:
 pass
-# clean up asyncio.
-self.disconnect()
 self.debugging = False
 # make sure we close down our end of the socket
 # might be overkill as normally stdin, stdout and stderr
 # SHOULD be closed on exit, but it does not hurt to do it here
 except SyntaxError:
 exctype, excval, exctb = sys.exc_info()
 try:
 message = str(excval)
 filename = excval.filename
-linenr = excval.lineno
+lineno = excval.lineno
-charnr = excval.offset
+charno = excval.offset
 except (AttributeError, ValueError):
-exclist = []
+message = ""
-else:
+filename = ""
-exclist = [message, [filename, linenr, charnr]]
+lineno = 0
+charno = 0
-self.write("{0}{1}\n".format(
+self.sendSyntaxError(message, filename, lineno, charno)
-DebugProtocol.ResponseSyntax, str(exclist)))
 return None
 return code
 def handleLine(self, line):
 # Remove any newline.
 if line[-1] == '\n':
 line = line[:-1]
 ##        printerr(line)          ##debug
-eoc = line.find('<')
+self.handleJsonCommand(line)
-if eoc >= 0 and line[0] == '>':
+def handleJsonCommand(self, jsonStr):
-# Get the command part and any argument.
+"""
-cmd = line[:eoc + 1]
+Public method to handle a command serialized as a JSON string.
-arg = line[eoc + 1:]
+@param jsonStr string containing the command received from the IDE
-if cmd == DebugProtocol.RequestVariables:
+@type str
-frmnr, scope, filter = eval(arg.replace("u'", "'"))
+"""
-self.__dumpVariables(int(frmnr), int(scope), filter)
+try:
-return
+commandDict = json.loads(jsonStr.strip())
+except (TypeError, ValueError) as err:
-if cmd == DebugProtocol.RequestVariable:
+printerr(str(err))
-var, frmnr, scope, filter = eval(arg.replace("u'", "'"))
+return
-self.__dumpVariable(var, int(frmnr), int(scope), filter)
-return
+method = commandDict["method"]
+params = commandDict["params"]
-if cmd == DebugProtocol.RequestThreadList:
-self.__dumpThreadList()
+if method == "RequestVariables":
-return
+self.__dumpVariables(
+params["frameNumber"], params["scope"], params["filters"])
-if cmd == DebugProtocol.RequestThreadSet:
-tid = eval(arg)
+elif method == "RequestVariable":
-if tid in self.threads:
+self.__dumpVariable(
-self.setCurrentThread(tid)
+params["variable"], params["frameNumber"],
-self.write(DebugProtocol.ResponseThreadSet + '\n')
+params["scope"], params["filters"])
-stack = self.currentThread.getStack()
-self.write('{0}{1!r}\n'.format(
+elif method == "RequestThreadList":
-DebugProtocol.ResponseStack, stack))
+self.__dumpThreadList()
-return
+elif method == "RequestThreadSet":
-if cmd == DebugProtocol.RequestStep:
+if params["threadID"] in self.threads:
-self.currentThread.step(True)
+self.setCurrentThread(params["threadID"])
-self.eventExit = True
+self.sendJsonCommand("ResponseThreadSet", {})
-return
+stack = self.currentThread.getStack()
+self.sendJsonCommand("ResponseStack", {
-if cmd == DebugProtocol.RequestStepOver:
+"stack": stack,
-self.currentThread.step(False)
+})
-self.eventExit = True
-return
+elif method == "RequestCapabilities":
+self.sendJsonCommand("ResponseCapabilities", {
-if cmd == DebugProtocol.RequestStepOut:
+"capabilities": self.__clientCapabilities(),
-self.currentThread.stepOut()
+"clientType": "Python3"
-self.eventExit = True
+})
-return
+elif method == "RequestBanner":
-if cmd == DebugProtocol.RequestStepQuit:
+self.sendJsonCommand("ResponseBanner", {
-if self.passive:
+"version": "Python {0}".format(sys.version),
-self.progTerminated(42)
+"platform": socket.gethostname(),
+"dbgclient": self.variant,
+})
+elif method == "RequestSetFilter":
+self.__generateFilterObjects(params["scope"], params["filter"])
+elif method == "RequestCallTrace":
+if params["enable"]:
+callTraceEnabled = self.profile
+else:
+callTraceEnabled = None
+if self.debugging:
+sys.setprofile(callTraceEnabled)
+else:
+# remember for later
+self.callTraceEnabled = callTraceEnabled
+elif method == "RequestEnvironment":
+for key, value in params["environment"].items():
+if key.endswith("+"):
+if key[:-1] in os.environ:
+os.environ[key[:-1]] += value
+else:
+os.environ[key[:-1]] = value
 else:
-self.set_quit()
+os.environ[key] = value
-self.eventExit = True
-return
+elif method == "RequestLoad":
+self._fncache = {}
-if cmd == DebugProtocol.RequestContinue:
+self.dircache = []
-special = int(arg)
+sys.argv = []
-self.currentThread.go(special)
+self.__setCoding(params["filename"])
-self.eventExit = True
+sys.argv.append(params["filename"])
-return
+sys.argv.extend(params["argv"])
+sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
-if cmd == DebugProtocol.RequestOK:
+if params["workdir"] == '':
-self.write(self.pendingResponse + '\n')
+os.chdir(sys.path[1])
-self.pendingResponse = DebugProtocol.ResponseOK
+else:
-return
+os.chdir(params["workdir"])
-if cmd == DebugProtocol.RequestCallTrace:
+self.running = sys.argv[0]
-if arg.strip().lower() == "on":
+self.mainFrame = None
-callTraceEnabled = self.profile
+self.debugging = True
+self.fork_auto = params["autofork"]
+self.fork_child = params["forkChild"]
+self.threads.clear()
+self.attachThread(mainThread=True)
+# set the system exception handling function to ensure, that
+# we report on all unhandled exceptions
+sys.excepthook = self.__unhandled_exception
+self.__interceptSignals()
+# clear all old breakpoints, they'll get set after we have
+# started
+Breakpoint.clear_all_breaks()
+Watch.clear_all_watches()
+self.mainThread.tracePythonLibs(params["traceInterpreter"])
+# This will eventually enter a local event loop.
+self.debugMod.__dict__['__file__'] = self.running
+sys.modules['__main__'] = self.debugMod
+code = self.__compileFileSource(self.running)
+if code:
+sys.setprofile(self.callTraceEnabled)
+res = self.mainThread.run(code, self.debugMod.__dict__)
+self.progTerminated(res)
+elif method == "RequestRun":
+sys.argv = []
+self.__setCoding(params["filename"])
+sys.argv.append(params["filename"])
+sys.argv.extend(params["argv"])
+sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
+if params["workdir"] == '':
+os.chdir(sys.path[1])
+else:
+os.chdir(params["workdir"])
+self.running = sys.argv[0]
+self.mainFrame = None
+self.botframe = None
+self.fork_auto = params["autofork"]
+self.fork_child = params["forkChild"]
+self.threads.clear()
+self.attachThread(mainThread=True)
+# set the system exception handling function to ensure, that
+# we report on all unhandled exceptions
+sys.excepthook = self.__unhandled_exception
+self.__interceptSignals()
+self.mainThread.tracePythonLibs(False)
+self.debugMod.__dict__['__file__'] = sys.argv[0]
+sys.modules['__main__'] = self.debugMod
+res = 0
+code = self.__compileFileSource(self.running)
+if code:
+try:
+exec(code, self.debugMod.__dict__)
+except SystemExit as exc:
+res = exc.code
+atexit._run_exitfuncs()
+self.writestream.flush()
+self.progTerminated(res)
+elif method == "RequestCoverage":
+from coverage import coverage
+sys.argv = []
+self.__setCoding(params["filename"])
+sys.argv.append(params["filename"])
+sys.argv.extend(params["argv"])
+sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
+if params["workdir"] == '':
+os.chdir(sys.path[1])
+else:
+os.chdir(params["workdir"])
+# set the system exception handling function to ensure, that
+# we report on all unhandled exceptions
+sys.excepthook = self.__unhandled_exception
+self.__interceptSignals()
+# generate a coverage object
+self.cover = coverage(
+auto_data=True,
+data_file="{0}.coverage".format(
+os.path.splitext(sys.argv[0])[0]))
+if params["erase"]:
+self.cover.erase()
+sys.modules['__main__'] = self.debugMod
+self.debugMod.__dict__['__file__'] = sys.argv[0]
+fp = open(sys.argv[0], encoding=self.__coding)
+try:
+script = fp.read()
+finally:
+fp.close()
+if script:
+if not script.endswith('\n'):
+script += '\n'
+code = compile(script, sys.argv[0], 'exec')
+self.running = sys.argv[0]
+res = 0
+self.cover.start()
+try:
+exec(code, self.debugMod.__dict__)
+except SystemExit as exc:
+res = exc.code
+atexit._run_exitfuncs()
+self.cover.stop()
+self.cover.save()
+self.writestream.flush()
+self.progTerminated(res)
+elif method == "RequestProfile":
+sys.setprofile(None)
+import PyProfile
+sys.argv = []
+self.__setCoding(params["filename"])
+sys.argv.append(params["filename"])
+sys.argv.extend(params["argv"])
+sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
+if params["workdir"] == '':
+os.chdir(sys.path[1])
+else:
+os.chdir(params["workdir"])
+# set the system exception handling function to ensure, that
+# we report on all unhandled exceptions
+sys.excepthook = self.__unhandled_exception
+self.__interceptSignals()
+# generate a profile object
+self.prof = PyProfile.PyProfile(sys.argv[0])
+if params["erase"]:
+self.prof.erase()
+self.debugMod.__dict__['__file__'] = sys.argv[0]
+sys.modules['__main__'] = self.debugMod
+fp = open(sys.argv[0], encoding=self.__coding)
+try:
+script = fp.read()
+finally:
+fp.close()
+if script:
+if not script.endswith('\n'):
+script += '\n'
+self.running = sys.argv[0]
+res = 0
+try:
+self.prof.run(script)
+except SystemExit as exc:
+res = exc.code
+atexit._run_exitfuncs()
+self.prof.save()
+self.writestream.flush()
+self.progTerminated(res)
+elif method == "ExecuteStatement":
+if self.buffer:
+self.buffer = self.buffer + '\n' + params["statement"]
+else:
+self.buffer = params["statement"]
+try:
+code = self.compile_command(self.buffer, self.readstream.name)
+except (OverflowError, SyntaxError, ValueError):
+# Report the exception
+sys.last_type, sys.last_value, sys.last_traceback = \
+sys.exc_info()
+self.sendJsonCommand("ClientOutput", {
+"text": "".join(traceback.format_exception_only(
+sys.last_type, sys.last_value))
+})
+self.buffer = ''
+else:
+if code is None:
+self.sendJsonCommand("ResponseContinue", {})
+return
 else:
-callTraceEnabled = None
+self.buffer = ''
-if self.debugging:
-sys.setprofile(callTraceEnabled)
+try:
-else:
+if self.running is None:
-# remember for later
+exec(code, self.debugMod.__dict__)
-self.callTraceEnabled = callTraceEnabled
-return
-if cmd == DebugProtocol.RequestEnv:
-env = eval(arg.replace("u'", "'"))
-for key, value in env.items():
-if key.endswith("+"):
-if key[:-1] in os.environ:
-os.environ[key[:-1]] += value
 else:
-os.environ[key[:-1]] = value
+if self.currentThread is None:
-else:
+# program has terminated
-os.environ[key] = value
-return
-if cmd == DebugProtocol.RequestLoad:
-self._fncache = {}
-self.dircache = []
-sys.argv = []
-wd, fn, args, tracePython = arg.split('|')
-self.__setCoding(fn)
-sys.argv.append(fn)
-sys.argv.extend(eval(args.replace("u'", "'")))
-sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
-if wd == '':
-os.chdir(sys.path[1])
-else:
-os.chdir(wd)
-self.running = sys.argv[0]
-self.mainFrame = None
-self.inRawMode = False
-self.debugging = True
-self.threads.clear()
-self.attachThread(mainThread=True)
-# set the system exception handling function to ensure, that
-# we report on all unhandled exceptions
-sys.excepthook = self.__unhandled_exception
-self.__interceptSignals()
-# clear all old breakpoints, they'll get set after we have
-# started
-Breakpoint.clear_all_breaks()
-Watch.clear_all_watches()
-self.mainThread.tracePythonLibs(int(tracePython))
-# This will eventually enter a local event loop.
-# Note the use of backquotes to cause a repr of self.running.
-# The need for this is on Windows os where backslash is the
-# path separator. They will get inadvertantly stripped away
-# during the eval causing IOErrors, if self.running is passed
-# as a normal str.
-self.debugMod.__dict__['__file__'] = self.running
-sys.modules['__main__'] = self.debugMod
-code = self.__compileFileSource(self.running)
-if code:
-sys.setprofile(self.callTraceEnabled)
-res = self.mainThread.run(code, self.debugMod.__dict__)
-self.progTerminated(res)
-return
-if cmd == DebugProtocol.RequestRun:
-sys.argv = []
-wd, fn, args = arg.split('|')
-self.__setCoding(fn)
-sys.argv.append(fn)
-sys.argv.extend(eval(args.replace("u'", "'")))
-sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
-if wd == '':
-os.chdir(sys.path[1])
-else:
-os.chdir(wd)
-self.running = sys.argv[0]
-self.mainFrame = None
-self.botframe = None
-self.inRawMode = False
-self.threads.clear()
-self.attachThread(mainThread=True)
-# set the system exception handling function to ensure, that
-# we report on all unhandled exceptions
-sys.excepthook = self.__unhandled_exception
-self.__interceptSignals()
-self.mainThread.tracePythonLibs(0)
-self.debugMod.__dict__['__file__'] = sys.argv[0]
-sys.modules['__main__'] = self.debugMod
-res = 0
-code = self.__compileFileSource(self.running)
-if code:
-try:
-exec(code, self.debugMod.__dict__)
-except SystemExit as exc:
-res = exc.code
-atexit._run_exitfuncs()
-self.writestream.flush()
-self.progTerminated(res)
-return
-if cmd == DebugProtocol.RequestProfile:
-sys.setprofile(None)
-import PyProfile
-sys.argv = []
-wd, fn, args, erase = arg.split('|')
-self.__setCoding(fn)
-sys.argv.append(fn)
-sys.argv.extend(eval(args.replace("u'", "'")))
-sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
-if wd == '':
-os.chdir(sys.path[1])
-else:
-os.chdir(wd)
-# set the system exception handling function to ensure, that
-# we report on all unhandled exceptions
-sys.excepthook = self.__unhandled_exception
-self.__interceptSignals()
-# generate a profile object
-self.prof = PyProfile.PyProfile(sys.argv[0])
-if int(erase):
-self.prof.erase()
-self.debugMod.__dict__['__file__'] = sys.argv[0]
-sys.modules['__main__'] = self.debugMod
-fp = open(sys.argv[0], encoding=self.__coding)
-try:
-script = fp.read()
-finally:
-fp.close()
-if script:
-if not script.endswith('\n'):
-script += '\n'
-self.running = sys.argv[0]
-res = 0
-try:
-self.prof.run(script)
-except SystemExit as exc:
-res = exc.code
-atexit._run_exitfuncs()
-self.prof.save()
-self.writestream.flush()
-self.progTerminated(res)
-return
-if cmd == DebugProtocol.RequestCoverage:
-from coverage import coverage
-sys.argv = []
-wd, fn, args, erase = arg.split('@@')
-self.__setCoding(fn)
-sys.argv.append(fn)
-sys.argv.extend(eval(args.replace("u'", "'")))
-sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
-if wd == '':
-os.chdir(sys.path[1])
-else:
-os.chdir(wd)
-# set the system exception handling function to ensure, that
-# we report on all unhandled exceptions
-sys.excepthook = self.__unhandled_exception
-self.__interceptSignals()
-# generate a coverage object
-self.cover = coverage(
-auto_data=True,
-data_file="{0}.coverage".format(
-os.path.splitext(sys.argv[0])[0]))
-if int(erase):
-self.cover.erase()
-sys.modules['__main__'] = self.debugMod
-self.debugMod.__dict__['__file__'] = sys.argv[0]
-fp = open(sys.argv[0], encoding=self.__coding)
-try:
-script = fp.read()
-finally:
-fp.close()
-if script:
-if not script.endswith('\n'):
-script += '\n'
-code = compile(script, sys.argv[0], 'exec')
-self.running = sys.argv[0]
-res = 0
-self.cover.start()
-try:
-exec(code, self.debugMod.__dict__)
-except SystemExit as exc:
-res = exc.code
-atexit._run_exitfuncs()
-self.cover.stop()
-self.cover.save()
-self.writestream.flush()
-self.progTerminated(res)
-return
-if cmd == DebugProtocol.RequestShutdown:
-self.sessionClose()
-return
-if cmd == DebugProtocol.RequestBreak:
-fn, line, temporary, set, cond = arg.split('@@')
-line = int(line)
-set = int(set)
-temporary = int(temporary)
-if set:
-if cond == 'None' or cond == '':
-cond = None
-else:
-try:
-cond = compile(cond, '<string>', 'eval')
-except SyntaxError:
-self.write('{0}{1},{2:d}\n'.format(
-DebugProtocol.ResponseBPConditionError,
-fn, line))
-return
-Breakpoint(fn, line, temporary, cond)
-else:
-Breakpoint.clear_break(fn, line)
-return
-if cmd == DebugProtocol.RequestBreakEnable:
-fn, line, enable = arg.split(',')
-line = int(line)
-enable = int(enable)
-bp = Breakpoint.get_break(fn, line)
-if bp is not None:
-if enable:
-bp.enable()
-else:
-bp.disable()
-return
-if cmd == DebugProtocol.RequestBreakIgnore:
-fn, line, count = arg.split(',')
-line = int(line)
-count = int(count)
-bp = Breakpoint.get_break(fn, line)
-if bp is not None:
-bp.ignore = count
-return
-if cmd == DebugProtocol.RequestWatch:
-cond, temporary, set = arg.split('@@')
-set = int(set)
-temporary = int(temporary)
-if cond.endswith(('??created??', '??changed??')):
-compiledCond, flag = cond.split()
-else:
-compiledCond = cond
-flag = ''
-try:
-compiledCond = compile(compiledCond, '<string>', 'eval')
-except SyntaxError:
-self.write('{0}{1}\n'.format(
-DebugProtocol.ResponseWPConditionError, cond))
-return
-if set:
-Watch(cond, compiledCond, flag, temporary)
-else:
-Watch.clear_watch(cond)
-return
-if cmd == DebugProtocol.RequestWatchEnable:
-cond, enable = arg.split(',')
-enable = int(enable)
-bp = Watch.get_watch(cond)
-if bp is not None:
-if enable:
-bp.enable()
-else:
-bp.disable()
-return
-if cmd == DebugProtocol.RequestWatchIgnore:
-cond, count = arg.split(',')
-count = int(count)
-bp = Watch.get_watch(cond)
-if bp is not None:
-bp.ignore = count
-return
-if cmd == DebugProtocol.RequestEval:
-try:
-value = eval(
-arg,
-self.currentThread.getCurrentFrame().f_globals,
-self.currentThread.getFrameLocals(self.framenr))
-self.currentThread.storeFrameLocals(self.framenr)
-except Exception:
-# Report the exception and the traceback
-try:
-type, value, tb = sys.exc_info()
-sys.last_type = type
-sys.last_value = value
-sys.last_traceback = tb
-tblist = traceback.extract_tb(tb)
-del tblist[:1]
-list = traceback.format_list(tblist)
-if list:
-list.insert(0, "Traceback (innermost last):\n")
-list[len(list):] = \
-traceback.format_exception_only(type, value)
-finally:
-tblist = tb = None
-for l in list:
-self.write(l)
-self.write(DebugProtocol.ResponseException + '\n')
-else:
-self.write(str(value) + '\n')
-self.write(DebugProtocol.ResponseOK + '\n')
-return
-if cmd == DebugProtocol.RequestExec:
-_globals = self.currentThread.getCurrentFrame().f_globals
-_locals = self.currentThread.getFrameLocals(self.framenr)
-try:
-code = compile(arg + '\n', '<stdin>', 'single')
-exec(code, _globals, _locals)
-self.currentThread.storeFrameLocals(self.framenr)
-except Exception:
-# Report the exception and the traceback
-try:
-type, value, tb = sys.exc_info()
-sys.last_type = type
-sys.last_value = value
-sys.last_traceback = tb
-tblist = traceback.extract_tb(tb)
-del tblist[:1]
-list = traceback.format_list(tblist)
-if list:
-list.insert(0, "Traceback (innermost last):\n")
-list[len(list):] = \
-traceback.format_exception_only(type, value)
-finally:
-tblist = tb = None
-for l in list:
-self.write(l)
-self.write(DebugProtocol.ResponseException + '\n')
-return
-if cmd == DebugProtocol.RequestBanner:
-self.write('{0}{1}\n'.format(DebugProtocol.ResponseBanner,
-str(("Python {0}".format(sys.version),
-socket.gethostname(), self.variant))))
-return
-if cmd == DebugProtocol.RequestCapabilities:
-self.write('{0}{1:d}, "Python3"\n'.format(
-DebugProtocol.ResponseCapabilities,
-self.__clientCapabilities()))
-return
-if cmd == DebugProtocol.RequestCompletion:
-self.__completionList(arg.replace("u'", "'"))
-return
-if cmd == DebugProtocol.RequestSetFilter:
-scope, filterString = eval(arg.replace("u'", "'"))
-self.__generateFilterObjects(int(scope), filterString)
-return
-if cmd == DebugProtocol.RequestUTPrepare:
-fn, tn, tfn, failed, cov, covname, erase = arg.split('|')
-sys.path.insert(0, os.path.dirname(os.path.abspath(fn)))
-os.chdir(sys.path[0])
-failed = eval(failed)
-# set the system exception handling function to ensure, that
-# we report on all unhandled exceptions
-sys.excepthook = self.__unhandled_exception
-self.__interceptSignals()
-try:
-import unittest
-utModule = imp.load_source(tn, fn)
-try:
-if failed:
-self.test = unittest.defaultTestLoader\
-.loadTestsFromNames(failed, utModule)
-else:
-self.test = unittest.defaultTestLoader\
-.loadTestsFromName(tfn, utModule)
-except AttributeError:
-self.test = unittest.defaultTestLoader\
-.loadTestsFromModule(utModule)
-except Exception:
-exc_type, exc_value, exc_tb = sys.exc_info()
-self.write('{0}{1}\n'.format(
-DebugProtocol.ResponseUTPrepared,
-str((0, str(exc_type), str(exc_value)))))
-self.__exceptionRaised()
-return
-# generate a coverage object
-if int(cov):
-from coverage import coverage
-self.cover = coverage(
-auto_data=True,
-data_file="{0}.coverage".format(
-os.path.splitext(covname)[0]))
-if int(erase):
-self.cover.erase()
-else:
-self.cover = None
-self.write('{0}{1}\n'.format(
-DebugProtocol.ResponseUTPrepared,
-str((self.test.countTestCases(), "", ""))))
-return
-if cmd == DebugProtocol.RequestUTRun:
-from DCTestResult import DCTestResult
-self.testResult = DCTestResult(self)
-if self.cover:
-self.cover.start()
-self.test.run(self.testResult)
-if self.cover:
-self.cover.stop()
-self.cover.save()
-self.write('{0}\n'.format(DebugProtocol.ResponseUTFinished))
-return
-if cmd == DebugProtocol.RequestUTStop:
-self.testResult.stop()
-return
-if cmd == DebugProtocol.ResponseForkTo:
-# this results from a separate event loop
-self.fork_child = (arg == 'child')
-self.eventExit = True
-return
-if cmd == DebugProtocol.RequestForkMode:
-self.fork_auto, self.fork_child = eval(arg)
-return
-# If we are handling raw mode input then reset the mode and break out
-# of the current event loop.
-if self.inRawMode:
-self.inRawMode = False
-self.rawLine = line
-self.eventExit = True
-return
-if self.buffer:
-self.buffer = self.buffer + '\n' + line
-else:
-self.buffer = line
-try:
-code = self.compile_command(self.buffer, self.readstream.name)
-except (OverflowError, SyntaxError, ValueError):
-# Report the exception
-sys.last_type, sys.last_value, sys.last_traceback = sys.exc_info()
-for l in traceback.format_exception_only(
-sys.last_type, sys.last_value):
-self.write(l)
-self.buffer = ''
-else:
-if code is None:
-self.pendingResponse = DebugProtocol.ResponseContinue
-else:
-self.buffer = ''
-try:
-if self.running is None:
-exec(code, self.debugMod.__dict__)
-else:
-if self.currentThread is None:
-# program has terminated
-self.running = None
-_globals = self.debugMod.__dict__
-_locals = _globals
-else:
-cf = self.currentThread.getCurrentFrame()
-# program has terminated
-if cf is None:
 self.running = None
 _globals = self.debugMod.__dict__
 _locals = _globals
 else:
-frmnr = self.framenr
+cf = self.currentThread.getCurrentFrame()
-while cf is not None and frmnr > 0:
+# program has terminated
-cf = cf.f_back
+if cf is None:
-frmnr -= 1
+self.running = None
-_globals = cf.f_globals
+_globals = self.debugMod.__dict__
-_locals = \
+_locals = _globals
-self.currentThread.getFrameLocals(
+else:
-self.framenr)
+frmnr = self.framenr
-# reset sys.stdout to our redirector (unconditionally)
+while cf is not None and frmnr > 0:
-if "sys" in _globals:
+cf = cf.f_back
-__stdout = _globals["sys"].stdout
+frmnr -= 1
-_globals["sys"].stdout = self.writestream
+_globals = cf.f_globals
-exec(code, _globals, _locals)
+_locals = \
-_globals["sys"].stdout = __stdout
+self.currentThread.getFrameLocals(
-elif "sys" in _locals:
+self.framenr)
-__stdout = _locals["sys"].stdout
+# reset sys.stdout to our redirector
-_locals["sys"].stdout = self.writestream
+# (unconditionally)
-exec(code, _globals, _locals)
+if "sys" in _globals:
-_locals["sys"].stdout = __stdout
+__stdout = _globals["sys"].stdout
-else:
+_globals["sys"].stdout = self.writestream
 exec(code, _globals, _locals)
+_globals["sys"].stdout = __stdout
-self.currentThread.storeFrameLocals(self.framenr)
+elif "sys" in _locals:
-except SystemExit as exc:
+__stdout = _locals["sys"].stdout
-self.progTerminated(exc.code)
+_locals["sys"].stdout = self.writestream
-except Exception:
+exec(code, _globals, _locals)
-# Report the exception and the traceback
+_locals["sys"].stdout = __stdout
+else:
+exec(code, _globals, _locals)
+self.currentThread.storeFrameLocals(self.framenr)
+except SystemExit as exc:
+self.progTerminated(exc.code)
+except Exception:
+# Report the exception and the traceback
+tlist = []
+try:
+exc_type, exc_value, exc_tb = sys.exc_info()
+sys.last_type = exc_type
+sys.last_value = exc_value
+sys.last_traceback = exc_tb
+tblist = traceback.extract_tb(exc_tb)
+del tblist[:1]
+tlist = traceback.format_list(tblist)
+if tlist:
+tlist.insert(
+0, "Traceback (innermost last):\n")
+tlist.extend(traceback.format_exception_only(
+exc_type, exc_value))
+finally:
+tblist = exc_tb = None
+self.sendJsonCommand("ClientOutput", {
+"text": "".join(tlist)
+})
+self.sendJsonCommand("ResponseOK", {})
+elif method == "RequestStep":
+self.currentThread.step(True)
+self.eventExit = True
+elif method == "RequestStepOver":
+self.currentThread.step(False)
+self.eventExit = True
+elif method == "RequestStepOut":
+self.currentThread.stepOut()
+self.eventExit = True
+elif method == "RequestStepQuit":
+if self.passive:
+self.progTerminated(42)
+else:
+self.set_quit()
+self.eventExit = True
+elif method == "RequestContinue":
+self.currentThread.go(params["special"])
+self.eventExit = True
+elif method == "RawInput":
+# If we are handling raw mode input then break out of the current
+# event loop.
+self.rawLine = params["input"]
+self.eventExit = True
+elif method == "RequestBreakpoint":
+if params["setBreakpoint"]:
+if params["condition"] in ['None', '']:
+cond = None
+elif params["condition"] is not None:
 try:
-exc_type, exc_value, exc_tb = sys.exc_info()
+cond = compile(params["condition"], '<string>', 'eval')
-sys.last_type = exc_type
+except SyntaxError:
-sys.last_value = exc_value
+self.sendJsonCommand("ResponseBPConditionError", {
-sys.last_traceback = exc_tb
+"filename": params["filename"],
-tblist = traceback.extract_tb(exc_tb)
+"line": params["line"],
-del tblist[:1]
+})
-list = traceback.format_list(tblist)
+return
-if list:
+else:
-list.insert(0, "Traceback (innermost last):\n")
+cond = None
-list[len(list):] = traceback.format_exception_only(
-exc_type, exc_value)
+Breakpoint(
-finally:
+params["filename"], params["line"], params["temporary"],
-tblist = exc_tb = None
+cond)
+else:
-for l in list:
+Breakpoint.clear_break(params["filename"], params["line"])
-self.write(l)
+elif method == "RequestBreakpointEnable":
+bp = Breakpoint.get_break(params["filename"], params["line"])
+if bp is not None:
+if params["enable"]:
+bp.enable()
+else:
+bp.disable()
+elif method == "RequestBreakpointIgnore":
+bp = Breakpoint.get_break(params["filename"], params["line"])
+if bp is not None:
+bp.ignore = params["count"]
+elif method == "RequestWatch":
+if params["setWatch"]:
+if params["condition"].endswith(
+('??created??', '??changed??')):
+compiledCond, flag = params["condition"].split()
+else:
+compiledCond = params["condition"]
+flag = ''
+try:
+compiledCond = compile(
+compiledCond, '<string>', 'eval')
+except SyntaxError:
+self.sendJsonCommand("ResponseWatchConditionError", {
+"condition": params["condition"],
+})
+return
+Watch(
+params["condition"], compiledCond, flag,
+params["temporary"])
+else:
+Watch.clear_watch(params["condition"])
+elif method == "RequestWatchEnable":
+wp = Watch.get_watch(params["condition"])
+if wp is not None:
+if params["enable"]:
+wp.enable()
+else:
+wp.disable()
+elif method == "RequestWatchIgnore":
+wp = Watch.get_watch(params["condition"])
+if wp is not None:
+wp.ignore = params["count"]
+elif method == "RequestShutdown":
+self.sessionClose()
+elif method == "RequestCompletion":
+self.__completionList(params["text"])
+elif method == "RequestUTPrepare":
+sys.path.insert(
+0, os.path.dirname(os.path.abspath(params["filename"])))
+os.chdir(sys.path[0])
+# set the system exception handling function to ensure, that
+# we report on all unhandled exceptions
+sys.excepthook = self.__unhandled_exception
+self.__interceptSignals()
+try:
+import unittest
+utModule = imp.load_source(
+params["testname"], params["filename"])
+try:
+if params["failed"]:
+self.test = unittest.defaultTestLoader\
+.loadTestsFromNames(params["failed"], utModule)
+else:
+self.test = unittest.defaultTestLoader\
+.loadTestsFromName(params["testfunctionname"],
+utModule)
+except AttributeError:
+self.test = unittest.defaultTestLoader\
+.loadTestsFromModule(utModule)
+except Exception:
+exc_type, exc_value, exc_tb = sys.exc_info()
+self.sendJsonCommand("ResponseUTPrepared", {
+"count": 0,
+"exception": exc_type.__name__,
+"message": str(exc_value),
+})
+return
+# generate a coverage object
+if params["coverage"]:
+from coverage import coverage
+self.cover = coverage(
+auto_data=True,
+data_file="{0}.coverage".format(
+os.path.splitext(params["coveragefile"])[0]))
+if params["coverageerase"]:
+self.cover.erase()
+else:
+self.cover = None
+self.sendJsonCommand("ResponseUTPrepared", {
+"count": self.test.countTestCases(),
+"exception": "",
+"message": "",
+})
+elif method == "RequestUTRun":
+from DCTestResult import DCTestResult
+self.testResult = DCTestResult(self)
+if self.cover:
+self.cover.start()
+self.test.run(self.testResult)
+if self.cover:
+self.cover.stop()
+self.cover.save()
+self.sendJsonCommand("ResponseUTFinished", {})
+elif method == "RequestUTStop":
+self.testResult.stop()
+elif method == "ResponseForkTo":
+# this results from a separate event loop
+self.fork_child = (params["target"] == 'child')
+self.eventExit = True
+def sendJsonCommand(self, method, params):
+"""
+Public method to send a single command or response to the IDE.
+@param method command or response command name to be sent
+@type str
+@param params dictionary of named parameters for the command or
+response
+@type dict
+"""
+cmd = prepareJsonCommand(method, params)
+self.writestream.write_p(cmd)
+self.writestream.flush()
+def sendClearTemporaryBreakpoint(self, filename, lineno):
+"""
+Public method to signal the deletion of a temporary breakpoint.
+@param filename name of the file the bp belongs to
+@type str
+@param lineno linenumber of the bp
+@type int
+"""
+self.sendJsonCommand("ResponseClearBreakpoint", {
+"filename": filename,
+"line": lineno
+})
+def sendClearTemporaryWatch(self, condition):
+"""
+Public method to signal the deletion of a temporary watch expression.
+@param condition condition of the watch expression to be cleared
+@type str
+"""
+self.sendJsonCommand("ResponseClearWatch", {
+"condition": condition,
+})
+def sendResponseLine(self, stack):
+"""
+Public method to send the current call stack.
+@param stack call stack
+@type list
+"""
+self.sendJsonCommand("ResponseLine", {
+"stack": stack,
+})
+def sendCallTrace(self, event, fromInfo, toInfo):
+"""
+Public method to send a call trace entry.
+@param event trace event (call or return)
+@type str
+@param fromInfo dictionary containing the origin info
+@type dict with 'filename', 'linenumber' and 'codename'
+as keys
+@param toInfo dictionary containing the target info
+@type dict with 'filename', 'linenumber' and 'codename'
+as keys
+"""
+self.sendJsonCommand("CallTrace", {
+"event": event[0],
+"from": fromInfo,
+"to": toInfo,
+})
+def sendException(self, exceptionType, exceptionMessage, stack):
+"""
+Public method to send information for an exception.
+@param exceptionType type of exception raised
+@type str
+@param exceptionMessage message of the exception
+@type str
+@param stack stack trace information
+@type list
+"""
+self.sendJsonCommand("ResponseException", {
+"type": exceptionType,
+"message": exceptionMessage,
+"stack": stack,
+})
+def sendSyntaxError(self, message, filename, lineno, charno):
+"""
+Public method to send information for a syntax error.
+@param message syntax error message
+@type str
+@param filename name of the faulty file
+@type str
+@param lineno line number info
+@type int
+@param charno character number info
+@type int
+"""
+self.sendJsonCommand("ResponseSyntax", {
+"message": message,
+"filename": filename,
+"linenumber": lineno,
+"characternumber": charno,
+})
+def sendPassiveStartup(self, filename, exceptions):
+"""
+Public method to send the passive start information.
+@param filename name of the script
+@type str
+@param exceptions flag to enable exception reporting of the IDE
+@type bool
+"""
+self.sendJsonCommand("PassiveStartup", {
+"filename": filename,
+"exceptions": exceptions,
+})
 def __clientCapabilities(self):
 """
 Private method to determine the clients capabilities.
 @return client capabilities (integer)
 return self.clientCapabilities
 except ImportError:
 return (
 self.clientCapabilities & ~DebugClientCapabilities.HasProfiler)
-def write(self, s):
+def readReady(self, stream):
 """
-Public method to write data to the output stream.
+Public method called when there is data ready to be read.
-@param s data to be written (string)
+@param stream file like object that has data to be written
 """
-self.writestream.write(s)
+try:
-self.writestream.flush()
+got = stream.readline_p()
+except Exception:
+return
+if len(got) == 0:
+self.sessionClose()
+return
+self.__receiveBuffer = self.__receiveBuffer + got
+# Call handleLine for the line if it is complete.
+eol = self.__receiveBuffer.find('\n')
+while eol >= 0:
+line = self.__receiveBuffer[:eol + 1]
+self.__receiveBuffer = self.__receiveBuffer[eol + 1:]
+self.handleLine(line)
+eol = self.__receiveBuffer.find('\n')
+def writeReady(self, stream):
+"""
+Public method called when we are ready to write data.
+@param stream file like object that has data to be written
+"""
+stream.write_p("")
+stream.flush()
 def __interact(self):
 """
 Private method to interact with the debugger.
 """
 global DebugClientInstance
-self.setDescriptors(self.readstream, self.writestream)
 DebugClientInstance = self
+self.__receiveBuffer = ""
 if not self.passive:
 # At this point simulate an event loop.
 self.eventLoop()
 except (select.error, KeyboardInterrupt, socket.error):
 # just carry on
 continue
 if self.readstream in rrdy:
-self.readReady(self.readstream.fileno())
+self.readReady(self.readstream)
 if self.writestream in wrdy:
-self.writeReady(self.writestream.fileno())
+self.writeReady(self.writestream)
 if self.errorstream in wrdy:
-self.writeReady(self.errorstream.fileno())
+self.writeReady(self.errorstream)
 self.eventExit = None
 self.pollingDisabled = False
 def eventPoll(self):
 rrdy, wrdy, xrdy = select.select([self.readstream], wrdy, [], 0)
 except (select.error, KeyboardInterrupt, socket.error):
 return
 if self.readstream in rrdy:
-self.readReady(self.readstream.fileno())
+self.readReady(self.readstream)
 if self.writestream in wrdy:
-self.writeReady(self.writestream.fileno())
+self.writeReady(self.writestream)
 if self.errorstream in wrdy:
-self.writeReady(self.errorstream.fileno())
+self.writeReady(self.errorstream)
 def connectDebugger(self, port, remoteAddress=None, redirect=True):
 """
 Public method to establish a session with the debugger.
 except Exception:
 fargs = ""
 else:
 fargs = ""
-siglist = [message, [filename, linenr, ffunc, fargs]]
+self.sendJsonCommand("ResponseSignal", {
+"message": message,
-self.write("{0}{1}".format(DebugProtocol.ResponseSignal, str(siglist)))
+"filename": filename,
+"linenumber": linenr,
+"function": ffunc,
+"arguments": fargs,
+})
 def absPath(self, fn):
 """
 Public method to convert a filename to an absolute name.
 @return flag indicating a running debug session (boolean)
 """
 return self.running
-def progTerminated(self, status):
+def progTerminated(self, status, message=""):
 """
 Public method to tell the debugger that the program has terminated.
 @param status return status
 @type int
+@param message status message
+@type str
 """
 if status is None:
 status = 0
-else:
+elif not isinstance(status, int):
-try:
+message = str(status)
-int(status)
+status = 1
-except ValueError:
-status = 1
 if self.running:
 self.set_quit()
 self.running = None
-self.write('{0}{1:d}\n'.format(DebugProtocol.ResponseExit, status))
+self.sendJsonCommand("ResponseExit", {
+"status": status,
+"message": message,
+})
 # reset coding
 self.__coding = self.defaultCoding
 def __dumpVariables(self, frmnr, scope, filter):
 elif f.f_globals is f.f_locals:
 scope = -1
 else:
 dict = f.f_locals
-varlist = [scope]
+varlist = []
 if scope != -1:
 keylist = dict.keys()
 vlist = self.__formatVariablesList(keylist, dict, scope, filter)
 varlist.extend(vlist)
-self.write('{0}{1}\n'.format(
+self.sendJsonCommand("ResponseVariables", {
-DebugProtocol.ResponseVariables, str(varlist)))
+"scope": scope,
+"variables": varlist,
+})
 def __dumpVariable(self, var, frmnr, scope, filter):
 """
 Private method to return the variables of a frame to the debug server.
 elif f.f_globals is f.f_locals:
 scope = -1
 else:
 dict = f.f_locals
-varlist = [scope, var]
+varlist = []
 if scope != -1:
-# search the correct dictionary
+variable = dict
-i = 0
+for attribute in var:
-rvar = var[:]
+attribute = self.__extractIndicators(attribute)[0]
-dictkeys = None
+typeObject, typeName, typeStr, resolver = \
-obj = None
+DebugVariables.getType(variable)
-isDict = False
+if resolver:
-formatSequences = False
+variable = resolver.resolve(variable, attribute)
-access = ""
-oaccess = ""
-odict = dict
-qtVariable = False
-qvar = None
-qvtype = ""
-while i < len(var):
-if len(dict):
-udict = dict
-ndict = {}
-# this has to be in line with VariablesViewer.indicators
-if var[i][-2:] in ["[]", "()", "{}"]:       # __IGNORE_WARNING__
-if i + 1 == len(var):
-if var[i][:-2] == '...':
-dictkeys = [var[i - 1]]
-else:
-dictkeys = [var[i][:-2]]
-formatSequences = True
-if not access and not oaccess:
-if var[i][:-2] == '...':
-access = '["{0!s}"]'.format(var[i - 1])
-dict = odict
-else:
-access = '["{0!s}"]'.format(var[i][:-2])
-else:
-if var[i][:-2] == '...':
-if oaccess:
-access = oaccess
-else:
-access = '{0!s}[{1!s}]'.format(
-access, var[i - 1])
-dict = odict
-else:
-if oaccess:
-access = '{0!s}[{1!s}]'.format(
-oaccess, var[i][:-2])
-oaccess = ''
-else:
-access = '{0!s}[{1!s}]'.format(
-access, var[i][:-2])
-if var[i][-2:] == "{}":     # __IGNORE_WARNING__
-isDict = True
-break
-else:
-if not access:
-if var[i][:-2] == '...':
-access = '["{0!s}"]'.format(var[i - 1])
-dict = odict
-else:
-access = '["{0!s}"]'.format(var[i][:-2])
-else:
-if var[i][:-2] == '...':
-access = '{0!s}[{1!s}]'.format(
-access, var[i - 1])
-dict = odict
-else:
-if oaccess:
-access = '{0!s}[{1!s}]'.format(
-oaccess, var[i][:-2])
-oaccess = ''
-else:
-access = '{0!s}[{1!s}]'.format(
-access, var[i][:-2])
 else:
-if access:
+break
-if oaccess:
+typeObject, typeName, typeStr, resolver = \
-access = '{0!s}[{1!s}]'.format(oaccess, var[i])
+DebugVariables.getType(variable)
-else:
+if typeStr.startswith(("PyQt5.", "PyQt4.")):
-access = '{0!s}[{1!s}]'.format(access, var[i])
+vlist = self.__formatQtVariable(variable, typeName)
-if var[i - 1][:-2] == '...':
+varlist.extend(vlist)
-oaccess = access
+elif resolver:
-else:
+dict = resolver.getDictionary(variable)
-oaccess = ''
+vlist = self.__formatVariablesList(
-try:
+list(dict.keys()), dict, scope, filter)
-loc = {"dict": dict}
+varlist.extend(vlist)
-exec('mdict = dict{0!s}.__dict__\nobj = dict{0!s}'
-.format(access), globals(), loc)
+self.sendJsonCommand("ResponseVariable", {
-mdict = loc["mdict"]
+"scope": scope,
-obj = loc["obj"]
+"variable": var,
-if "PyQt4." in str(type(obj)) or \
+"variables": varlist,
-"PyQt5." in str(type(obj)):
+})
-qtVariable = True
-qvar = obj
+def __extractIndicators(self, var):
-qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
+"""
-ndict.update(mdict)
+Private method to extract the indicator string from a variable text.
-except Exception:
-pass
+@param var variable text
-try:
+@type str
-loc = {"dict": dict}
+@return tuple containing the variable text without indicators and the
-exec('mcdict = dict{0!s}.__class__.__dict__'
+indicator string
-.format(access), globals(), loc)
+@rtype tuple of two str
-ndict.update(loc["mcdict"])
+"""
-if mdict and "sipThis" not in mdict.keys():
+for indicator in DebugClientBase.Indicators:
-del rvar[0:2]
+if var.endswith(indicator):
-access = ""
+return var[:-len(indicator)], indicator
-except Exception:
-pass
+return var, ""
-try:
-loc = {"cdict": {}, "dict": dict}
+def __formatQtVariable(self, value, qttype):
-exec('slv = dict{0!s}.__slots__'.format(access),
-globals(), loc)
-for v in loc["slv"]:
-try:
-loc["v"] = v
-exec('cdict[v] = dict{0!s}.{1!s}'.format(
-access, v), globals, loc)
-except Exception:
-pass
-ndict.update(loc["cdict"])
-exec('obj = dict{0!s}'.format(access),
-globals(), loc)
-obj = loc["obj"]
-access = ""
-if "PyQt4." in str(type(obj)) or \
-"PyQt5." in str(type(obj)):
-qtVariable = True
-qvar = obj
-qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
-except Exception:
-pass
-else:
-try:
-ndict.update(dict[var[i]].__dict__)
-ndict.update(dict[var[i]].__class__.__dict__)
-del rvar[0]
-obj = dict[var[i]]
-if "PyQt4." in str(type(obj)) or \
-"PyQt5." in str(type(obj)):
-qtVariable = True
-qvar = obj
-qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
-except Exception:
-pass
-try:
-slv = dict[var[i]].__slots__
-loc = {"cdict": {}, "dict": dict,
-"var": var, "i": i}
-for v in slv:
-try:
-loc["v"] = v
-exec('cdict[v] = dict[var[i]].{0!s}'
-.format(v),
-globals(), loc)
-except Exception:
-pass
-ndict.update(loc["cdict"])
-obj = dict[var[i]]
-if "PyQt4." in str(type(obj)) or \
-"PyQt5." in str(type(obj)):
-qtVariable = True
-qvar = obj
-qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
-except Exception:
-pass
-odict = dict
-dict = ndict
-i += 1
-if qtVariable:
-vlist = self.__formatQtVariable(qvar, qvtype)
-elif ("sipThis" in dict.keys() and len(dict) == 1) or \
-(len(dict) == 0 and len(udict) > 0):
-if access:
-loc = {"udict": udict}
-exec('qvar = udict{0!s}'.format(access), globals(), loc)
-qvar = loc["qvar"]
-# this has to be in line with VariablesViewer.indicators
-elif rvar and rvar[0][-2:] in ["[]", "()", "{}"]:   # __IGNORE_WARNING__
-loc = {"udict": udict}
-exec('qvar = udict["{0!s}"][{1!s}]'.format(rvar[0][:-2],
-rvar[1]),
-globals(), loc)
-qvar = loc["qvar"]
-else:
-qvar = udict[var[-1]]
-qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
-if qvtype.startswith(("PyQt4", "PyQt5")):
-vlist = self.__formatQtVariable(qvar, qvtype)
-else:
-vlist = []
-else:
-qtVariable = False
-if len(dict) == 0 and len(udict) > 0:
-if access:
-loc = {"udict": udict}
-exec('qvar = udict{0!s}'.format(access),
-globals(), loc)
-qvar = loc["qvar"]
-# this has to be in line with VariablesViewer.indicators
-elif rvar and rvar[0][-2:] in ["[]", "()", "{}"]:   # __IGNORE_WARNING__
-loc = {"udict": udict}
-exec('qvar = udict["{0!s}"][{1!s}]'.format(
-rvar[0][:-2], rvar[1]), globals(), loc)
-qvar = loc["qvar"]
-else:
-qvar = udict[var[-1]]
-qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
-if qvtype.startswith(("PyQt4", "PyQt5")):
-qtVariable = True
-if qtVariable:
-vlist = self.__formatQtVariable(qvar, qvtype)
-else:
-# format the dictionary found
-if dictkeys is None:
-dictkeys = dict.keys()
-else:
-# treatment for sequences and dictionaries
-if access:
-loc = {"dict": dict}
-exec("dict = dict{0!s}".format(access), globals(),
-loc)
-dict = loc["dict"]
-else:
-dict = dict[dictkeys[0]]
-if isDict:
-dictkeys = dict.keys()
-else:
-dictkeys = range(len(dict))
-vlist = self.__formatVariablesList(
-dictkeys, dict, scope, filter, formatSequences)
-varlist.extend(vlist)
-if obj is not None and not formatSequences:
-try:
-if repr(obj).startswith('{'):
-varlist.append(
-('...', 'dict', "{0:d}".format(len(obj.keys()))))
-elif repr(obj).startswith('['):
-varlist.append(
-('...', 'list', "{0:d}".format(len(obj))))
-elif repr(obj).startswith('('):
-varlist.append(
-('...', 'tuple', "{0:d}".format(len(obj))))
-except Exception:
-pass
-self.write('{0}{1}\n'.format(
-DebugProtocol.ResponseVariable, str(varlist)))
-def __formatQtVariable(self, value, vtype):
 """
 Private method to produce a formatted output of a simple Qt4/Qt5 type.
 @param value variable to be formatted
-@param vtype type of the variable to be formatted (string)
+@param qttype type of the Qt variable to be formatted (string)
 @return A tuple consisting of a list of formatted variables. Each
 variable entry is a tuple of three elements, the variable name,
 its type and value.
 """
-qttype = vtype.split('.')[-1]
 varlist = []
 if qttype == 'QChar':
 varlist.append(("", "QChar", "{0}".format(chr(value.unicode()))))
 varlist.append(("", "int", "{0:d}".format(value.unicode())))
 elif qttype == 'QByteArray':
 break
 if matched:
 continue
 # filter hidden attributes (filter #0)
-if 0 in filter and str(key)[:2] == '__':
+if 0 in filter and str(key)[:2] == '__' and not (
+key == "___len___" and
+DebugVariables.TooLargeAttribute in keylist):
 continue
 # special handling for '__builtins__' (it's way too big)
 if key == '__builtins__':
 rvalue = '<module __builtin__ (built-in)>'
 if valtype not in ConfigVarTypeStrings:
 if ConfigVarTypeStrings.index('instance') in filter:
 continue
 elif valtype == "sip.methoddescriptor":
 if ConfigVarTypeStrings.index(
-'instance method') in filter:
+'method') in filter:
 continue
 elif valtype == "sip.enumtype":
 if ConfigVarTypeStrings.index('class') in filter:
 continue
 valtype = valtypestr
 if ConfigVarTypeStrings.index(
 'instance') in filter:
 continue
 elif valtype == "sip.methoddescriptor":
 if ConfigVarTypeStrings.index(
-'instance method') in filter:
+'method') in filter:
 continue
 elif valtype == "sip.enumtype":
 if ConfigVarTypeStrings.index('class') in filter:
 continue
 elif not valtype.startswith("PySide") and \
 ConfigVarTypeStrings.index('other') in filter:
 continue
 try:
-if valtype not in ['list', 'tuple', 'dict']:
+if valtype not in ['list', 'tuple', 'dict', 'set',
+'frozenset']:
 rvalue = repr(value)
 if valtype.startswith('class') and \
 rvalue[0] in ['{', '(', '[']:
 rvalue = ""
 else:
 self.__getCompletionList(text, localCompleter, completions)
 except AttributeError:
 pass
 self.__getCompletionList(text, self.complete, completions)
-self.write("{0}{1}||{2}\n".format(DebugProtocol.ResponseCompletion,
+self.sendJsonCommand("ResponseCompletion", {
-str(list(completions)), text))
+"completions": list(completions),
+"text": text,
+})
 def __getCompletionList(self, text, completer, completions):
 """
 Private method to create a completions list.
 except IndexError:
 self.running = None
 if self.running:
 self.__setCoding(self.running)
 self.passive = True
-self.write("{0}{1}|{2:d}\n".format(DebugProtocol.PassiveStartup,
+self.sendPassiveStartup(self.running, exceptions)
-self.running, exceptions))
 self.__interact()
 # setup the debugger variables
 self._fncache = {}
 self.dircache = []
 self.mainFrame = None
-self.inRawMode = False
 self.debugging = True
 self.attachThread(mainThread=True)
 self.mainThread.tracePythonLibs(tracePython)
 else:
 os.chdir(wd)
 self.running = sys.argv[0]
 self.__setCoding(self.running)
 self.mainFrame = None
-self.inRawMode = False
 self.debugging = True
 self.passive = True
-self.write("{0}{1}|{2:d}\n".format(
+self.sendPassiveStartup(self.running, exceptions)
-DebugProtocol.PassiveStartup, self.running, exceptions))
 self.__interact()
 self.attachThread(mainThread=True)
 self.mainThread.tracePythonLibs(tracePython)
 to follow.
 @return process ID (integer)
 """
 if not self.fork_auto:
-self.write(DebugProtocol.RequestForkTo + '\n')
+self.sendJsonCommand("RequestForkTo", {})
 self.eventLoop(True)
 pid = DebugClientOrigFork()
 if pid == 0:
 # child
 if not self.fork_child:

Mercurial Repositories > eric / file comparison

comparison: DebugClients/Python3/DebugClientBase.py

DebugClients/Python3/DebugClientBase.py