eric: DebugClients/Python3/DebugClientBase.py@684f5ba04f0b (annotated)

 # -*- coding: utf-8 -*-
 # Copyright (c) 2009 - 2016 Detlev Offenbach <detlev@die-offenbachs.de>
 #
 """
 Module implementing a debug client base class.
 """
 import sys
 import socket
 import select
 import codeop
 import traceback
 import os
 import time
 import imp
 import re
 import atexit
 import signal
 import DebugProtocol
 import DebugClientCapabilities
 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
 DebugClientInstance = None
 ###############################################################################
 def DebugClientInput(prompt="", echo=True):
     """
     Replacement for the standard input builtin.
     This function works with the split debugger.
     @param prompt prompt to be shown (string)
     @param echo flag indicating to echo the output (boolean)
     @return result of the input() call
     """
     if DebugClientInstance is None or not DebugClientInstance.redirect:
         return DebugClientOrigInput(prompt)
     return DebugClientInstance.input(prompt, echo)
 # Use our own input().
 try:
     DebugClientOrigInput = __builtins__.__dict__['input']
     __builtins__.__dict__['input'] = DebugClientInput
 except (AttributeError, KeyError):
     import __main__
     DebugClientOrigInput = __main__.__builtins__.__dict__['input']
     __main__.__builtins__.__dict__['input'] = DebugClientInput
 ###############################################################################
 def DebugClientFork():
     """
     Replacement for the standard os.fork().
     @return result of the fork() call
     """
     if DebugClientInstance is None:
         return DebugClientOrigFork()
     return DebugClientInstance.fork()
 # use our own fork().
 if 'fork' in dir(os):
     DebugClientOrigFork = os.fork
     os.fork = DebugClientFork
 ###############################################################################
 def DebugClientClose(fd):
     """
     Replacement for the standard os.close(fd).
     @param fd open file descriptor to be closed (integer)
     """
     if DebugClientInstance is None:
         DebugClientOrigClose(fd)
     DebugClientInstance.close(fd)
 # use our own close().
 if 'close' in dir(os):
     DebugClientOrigClose = os.close
     os.close = DebugClientClose
 ###############################################################################
 def DebugClientSetRecursionLimit(limit):
     """
     Replacement for the standard sys.setrecursionlimit(limit).
     @param limit recursion limit (integer)
     """
     rl = max(limit, 64)
     setRecursionLimit(rl)
     DebugClientOrigSetRecursionLimit(rl + 64)
 # use our own setrecursionlimit().
 if 'setrecursionlimit' in dir(sys):
     DebugClientOrigSetRecursionLimit = sys.setrecursionlimit
     sys.setrecursionlimit = DebugClientSetRecursionLimit
     DebugClientSetRecursionLimit(sys.getrecursionlimit())
 ###############################################################################
 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.
     The protocol between the debugger and the client assumes that there will be
     a single source of debugger commands and a single source of Python
     statements.  Commands and statement are always exactly one line and may be
     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
     def __init__(self):
         """
         Constructor
         """
         self.breakpoints = {}
         self.redirect = True
         # The next couple of members are needed for the threaded version.
         # For this base class they contain static values for the non threaded
         # debugger
         # dictionary of all threads running
         self.threads = {}
         # the "current" thread, basically the thread we are at a breakpoint
         # for.
         self.currentThread = self
         # special objects representing the main scripts thread and frame
         self.mainThread = self
         self.mainFrame = None
         self.framenr = 0
         # The context to run the debugged program in.
         self.debugMod = imp.new_module('__main__')
         self.debugMod.__dict__['__builtins__'] = __builtins__
         # The list of complete lines to execute.
         self.buffer = ''
         # 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.tracePython = False
         self.debugging = False
         self.fork_auto = False
         self.fork_child = False
         self.readstream = None
         self.writestream = None
         self.errorstream = None
         self.pollingDisabled = False
         self.callTraceEnabled = False
         self.__newCallTraceEnabled = False
         self.skipdirs = sys.path[:]
         self.variant = 'You should not see this'
         # commandline completion stuff
         self.complete = Completer(self.debugMod.__dict__).complete
         self.compile_command = codeop.CommandCompiler()
         self.coding_re = re.compile(r"coding[:=]\s*([-\w_.]+)")
         self.defaultCoding = 'utf-8'
         self.__coding = self.defaultCoding
         self.noencoding = False
     def getCoding(self):
         """
         Public method to return the current coding.
         @return codec name (string)
         """
         return self.__coding
     def __setCoding(self, filename):
         """
         Private method to set the coding used by a python file.
         @param filename name of the file to inspect (string)
         """
         if self.noencoding:
             self.__coding = sys.getdefaultencoding()
         else:
             default = 'utf-8'
             try:
                 f = open(filename, 'rb')
                 # read the first and second line
                 text = f.readline()
                 text = "{0}{1}".format(text, f.readline())
                 f.close()
             except IOError:
                 self.__coding = default
                 return
             for l in text.splitlines():
                 m = self.coding_re.search(l)
                 if m:
                     self.__coding = m.group(1)
                     return
             self.__coding = default
     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
         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
         @param kwargs keyword arguments to pass to target
         @param mainThread True, if we are attaching to the already
               started mainthread of the app
         """
         if self.debugging:
             sys.setprofile(self.profile)
     def __dumpThreadList(self):
         """
         Private method to send the list of threads.
         """
         threadList = []
         if self.threads and self.currentThread:
             # indication for the threaded debugger
             currentId = self.currentThread.get_ident()
             for t in self.threads.values():
                 d = {}
                 d["id"] = t.get_ident()
                 d["name"] = t.get_name()
                 d["broken"] = t.isBroken()
                 threadList.append(d)
         else:
             currentId = -1
             d = {}
             d["id"] = -1
             d["name"] = "MainThread"
             if hasattr(self, "isBroken"):
                 d["broken"] = self.isBroken()
             else:
                 d["broken"] = False
             threadList.append(d)
         self.write("{0}{1!r}\n".format(DebugProtocol.ResponseThreadList,
                                        (currentId, 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(
             DebugProtocol.ResponseRaw, (prompt, echo)))
         self.inRawMode = True
         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.
         @param exit flag indicating to terminate (boolean)
         """
         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
         self.readstream.close(True)
         self.writestream.close(True)
         self.errorstream.close(True)
         if exit:
             # Ok, go away.
             sys.exit()
     def __compileFileSource(self, filename, mode='exec'):
         """
         Private method to compile source code read from a file.
         @param filename name of the source file (string)
         @param mode kind of code to be generated (string, exec or eval)
         @return compiled code object (None in case of errors)
         """
         with open(filename, encoding=self.__coding) as fp:
             statement = fp.read()
         try:
             code = compile(statement + '\n', filename, mode)
         except SyntaxError:
             exctype, excval, exctb = sys.exc_info()
             try:
                 message = str(excval)
                 filename = excval.filename
                 linenr = excval.lineno
                 charnr = excval.offset
             except (AttributeError, ValueError):
                 exclist = []
             else:
                 exclist = [message, [filename, linenr, charnr]]
             self.write("{0}{1}\n".format(
                 DebugProtocol.ResponseSyntax, str(exclist)))
             return None
         return code
     def handleLine(self, line):
         """
         Public method to handle the receipt of a complete line.
         It first looks for a valid protocol token at the start of the line.
         Thereafter it trys to execute the lines accumulated so far.
         @param line the received line
         """
         # Remove any newline.
         if line[-1] == '\n':
             line = line[:-1]
 ##        printerr(line)          ##debug
         eoc = line.find('<')
         if eoc >= 0 and line[0] == '>':
             # Get the command part and any argument.
             cmd = line[:eoc + 1]
             arg = line[eoc + 1:]
             if cmd == DebugProtocol.RequestVariables:
                 frmnr, scope, filter = eval(arg.replace("u'", "'"))
                 self.__dumpVariables(int(frmnr), int(scope), filter)
                 return
             if cmd == DebugProtocol.RequestVariable:
                 var, frmnr, scope, filter = eval(arg.replace("u'", "'"))
                 self.__dumpVariable(var, int(frmnr), int(scope), filter)
                 return
             if cmd == DebugProtocol.RequestThreadList:
                 self.__dumpThreadList()
                 return
             if cmd == DebugProtocol.RequestThreadSet:
                 tid = eval(arg)
                 if tid in self.threads:
                     self.setCurrentThread(tid)
                     self.write(DebugProtocol.ResponseThreadSet + '\n')
                     stack = self.currentThread.getStack()
                     self.write('{0}{1!r}\n'.format(
                         DebugProtocol.ResponseStack, stack))
                 return
             if cmd == DebugProtocol.RequestStep:
                 self.currentThread.step(True)
                 self.eventExit = True
                 return
             if cmd == DebugProtocol.RequestStepOver:
                 self.currentThread.step(False)
                 self.eventExit = True
                 return
             if cmd == DebugProtocol.RequestStepOut:
                 self.currentThread.stepOut()
                 self.eventExit = True
                 return
             if cmd == DebugProtocol.RequestStepQuit:
                 if self.passive:
                     self.progTerminated(42)
                 else:
                     self.set_quit()
                     self.eventExit = True
                 return
             if cmd == DebugProtocol.RequestContinue:
                 special = int(arg)
                 self.currentThread.go(special)
                 self.eventExit = True
                 return
             if cmd == DebugProtocol.RequestOK:
                 self.write(self.pendingResponse + '\n')
                 self.pendingResponse = DebugProtocol.ResponseOK
                 return
             if cmd == DebugProtocol.RequestCallTrace:
                 if arg.strip().lower() == "on":
                     callTraceEnabled = True
                 else:
                     callTraceEnabled = False
                 if self.debugging:
                     self.callTraceEnabled = callTraceEnabled
                 else:
                     self.__newCallTraceEnabled = callTraceEnabled
                     # remember for later
                 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
                     else:
                         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
                 self.mainThread.clear_all_breaks()
                 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:
                     self.callTraceEnabled = self.__newCallTraceEnabled
                     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:
                             compile(cond, '<string>', 'eval')
                         except SyntaxError:
                             self.write('{0}{1},{2:d}\n'.format(
                                        DebugProtocol.ResponseBPConditionError,
                                        fn, line))
                             return
                     self.mainThread.set_break(fn, line, temporary, cond)
                 else:
                     self.mainThread.clear_break(fn, line)
                 return
             if cmd == DebugProtocol.RequestBreakEnable:
                 fn, line, enable = arg.split(',')
                 line = int(line)
                 enable = int(enable)
                 bp = self.mainThread.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 = self.mainThread.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 set:
                     if not cond.endswith('??created??') and \
                        not cond.endswith('??changed??'):
                         try:
                             compile(cond, '<string>', 'eval')
                         except SyntaxError:
                             self.write('{0}{1}\n'.format(
                                 DebugProtocol.ResponseWPConditionError, cond))
                             return
                     self.mainThread.set_watch(cond, temporary)
                 else:
                     self.mainThread.clear_watch(cond)
                 return
             if cmd == DebugProtocol.RequestWatchEnable:
                 cond, enable = arg.split(',')
                 enable = int(enable)
                 bp = self.mainThread.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 = self.mainThread.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
                                 while cf is not None and frmnr > 0:
                                     cf = cf.f_back
                                     frmnr -= 1
                                 _globals = cf.f_globals
                                 _locals = \
                                     self.currentThread.getFrameLocals(
                                         self.framenr)
                         # reset sys.stdout to our redirector (unconditionally)
                         if "sys" in _globals:
                             __stdout = _globals["sys"].stdout
                             _globals["sys"].stdout = self.writestream
                             exec(code, _globals, _locals)
                             _globals["sys"].stdout = __stdout
                         elif "sys" in _locals:
                             __stdout = _locals["sys"].stdout
                             _locals["sys"].stdout = self.writestream
                             exec(code, _globals, _locals)
                             _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
                     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]
                         list = traceback.format_list(tblist)
                         if list:
                             list.insert(0, "Traceback (innermost last):\n")
                             list[len(list):] = traceback.format_exception_only(
                                 exc_type, exc_value)
                     finally:
                         tblist = exc_tb = None
                     for l in list:
                         self.write(l)
     def __clientCapabilities(self):
         """
         Private method to determine the clients capabilities.
         @return client capabilities (integer)
         """
         try:
             import PyProfile    # __IGNORE_WARNING__
             try:
                 del sys.modules['PyProfile']
             except KeyError:
                 pass
             return self.clientCapabilities
         except ImportError:
             return (
                 self.clientCapabilities & ~DebugClientCapabilities.HasProfiler)
     def write(self, s):
         """
         Public method to write data to the output stream.
         @param s data to be written (string)
         """
         self.writestream.write(s)
         self.writestream.flush()
     def __interact(self):
         """
         Private method to interact with the debugger.
         """
         global DebugClientInstance
         self.setDescriptors(self.readstream, self.writestream)
         DebugClientInstance = self
         if not self.passive:
             # At this point simulate an event loop.
             self.eventLoop()
     def eventLoop(self, disablePolling=False):
         """
         Public method implementing our event loop.
         @param disablePolling flag indicating to enter an event loop with
             polling disabled (boolean)
         """
         self.eventExit = None
         self.pollingDisabled = disablePolling
         while self.eventExit is None:
             wrdy = []
             if self.writestream.nWriteErrors > self.writestream.maxtries:
                 break
             if AsyncPendingWrite(self.writestream):
                 wrdy.append(self.writestream)
             if AsyncPendingWrite(self.errorstream):
                 wrdy.append(self.errorstream)
             try:
                 rrdy, wrdy, xrdy = select.select([self.readstream], wrdy, [])
             except (select.error, KeyboardInterrupt, socket.error):
                 # just carry on
                 continue
             if self.readstream in rrdy:
                 self.readReady(self.readstream.fileno())
             if self.writestream in wrdy:
                 self.writeReady(self.writestream.fileno())
             if self.errorstream in wrdy:
                 self.writeReady(self.errorstream.fileno())
         self.eventExit = None
 de9c2

Mercurial Repositories > eric / annotate

DebugClients/Python3/DebugClientBase.py@684f5ba04f0b (annotated)

DebugClients/Python3/DebugClientBase.py