--- a/Debugger/DebuggerInterfacePython.py Sat Sep 03 18:01:19 2016 +0200
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,1105 +0,0 @@
-# -*- coding: utf-8 -*-
-
-# Copyright (c) 2007 - 2016 Detlev Offenbach <detlev@die-offenbachs.de>
-#
-
-"""
-Module implementing the Python debugger interface for the debug server.
-"""
-
-from __future__ import unicode_literals
-
-import sys
-import os
-import re
-
-from PyQt5.QtCore import QObject, QTextCodec, QProcess, QProcessEnvironment, \
- QTimer
-from PyQt5.QtWidgets import QInputDialog
-
-from E5Gui.E5Application import e5App
-from E5Gui import E5MessageBox
-
-from . import DebugProtocol
-from . import DebugClientCapabilities
-
-import Preferences
-import Utilities
-
-from eric6config import getConfig
-
-
-ClientDefaultCapabilities = DebugClientCapabilities.HasAll
-
-
-class DebuggerInterfacePython(QObject):
- """
- Class implementing the Python debugger interface for the debug server.
- """
- def __init__(self, debugServer, passive):
- """
- Constructor
-
- @param debugServer reference to the debug server (DebugServer)
- @param passive flag indicating passive connection mode (boolean)
- """
- super(DebuggerInterfacePython, self).__init__()
-
- self.__isNetworked = True
- self.__autoContinue = False
-
- self.debugServer = debugServer
- self.passive = passive
- self.process = None
-
- self.qsock = None
- self.queue = []
-
- # set default values for capabilities of clients
- self.clientCapabilities = ClientDefaultCapabilities
-
- # set translation function
- self.translate = self.__identityTranslation
-
- self.codec = QTextCodec.codecForName(
- Preferences.getSystem("StringEncoding"))
-
- self.__unicodeRe = re.compile(r"""\bu(["'])""")
-
- if passive:
- # set translation function
- if Preferences.getDebugger("PathTranslation"):
- self.translateRemote = \
- Preferences.getDebugger("PathTranslationRemote")
- self.translateLocal = \
- Preferences.getDebugger("PathTranslationLocal")
- self.translate = self.__remoteTranslation
- else:
- self.translate = self.__identityTranslation
-
- # attribute to remember the name of the executed script
- self.__scriptName = ""
-
- def __identityTranslation(self, fn, remote2local=True):
- """
- Private method to perform the identity path translation.
-
- @param fn filename to be translated (string)
- @param remote2local flag indicating the direction of translation
- (False = local to remote, True = remote to local [default])
- @return translated filename (string)
- """
- return fn
-
- def __remoteTranslation(self, fn, remote2local=True):
- """
- Private method to perform the path translation.
-
- @param fn filename to be translated (string)
- @param remote2local flag indicating the direction of translation
- (False = local to remote, True = remote to local [default])
- @return translated filename (string)
- """
- if remote2local:
- return fn.replace(self.translateRemote, self.translateLocal)
- else:
- return fn.replace(self.translateLocal, self.translateRemote)
-
- def __startProcess(self, program, arguments, environment=None):
- """
- Private method to start the debugger client process.
-
- @param program name of the executable to start (string)
- @param arguments arguments to be passed to the program (list of string)
- @param environment dictionary of environment settings to pass
- (dict of string)
- @return the process object (QProcess) or None
- """
- proc = QProcess()
- if environment is not None:
- env = QProcessEnvironment()
- for key, value in list(environment.items()):
- env.insert(key, value)
- proc.setProcessEnvironment(env)
- args = []
- for arg in arguments:
- args.append(arg)
- proc.start(program, args)
- if not proc.waitForStarted(10000):
- proc = None
-
- return proc
-
- def startRemote(self, port, runInConsole):
- """
- Public method to start a remote Python interpreter.
-
- @param port portnumber the debug server is listening on (integer)
- @param runInConsole flag indicating to start the debugger in a
- console window (boolean)
- @return client process object (QProcess), a flag to indicate
- a network connection (boolean) and the name of the interpreter
- in case of a local execution (string)
- """
- interpreter = Preferences.getDebugger("PythonInterpreter")
- if interpreter == "":
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>No Python2 interpreter configured.</p>"""))
- return None, False, ""
-
- debugClientType = Preferences.getDebugger("DebugClientType")
- if debugClientType == "standard":
- debugClient = os.path.join(getConfig('ericDir'),
- "DebugClients", "Python",
- "DebugClient.py")
- elif debugClientType == "threaded":
- debugClient = os.path.join(getConfig('ericDir'),
- "DebugClients", "Python",
- "DebugClientThreads.py")
- else:
- debugClient = Preferences.getDebugger("DebugClient")
- if debugClient == "":
- debugClient = os.path.join(sys.path[0],
- "DebugClients", "Python",
- "DebugClient.py")
-
- redirect = str(Preferences.getDebugger("PythonRedirect"))
- noencoding = Preferences.getDebugger("PythonNoEncoding") and \
- '--no-encoding' or ''
-
- if Preferences.getDebugger("RemoteDbgEnabled"):
- ipaddr = self.debugServer.getHostAddress(False)
- rexec = Preferences.getDebugger("RemoteExecution")
- rhost = Preferences.getDebugger("RemoteHost")
- if rhost == "":
- rhost = "localhost"
- if rexec:
- args = Utilities.parseOptionString(rexec) + \
- [rhost, interpreter, debugClient,
- noencoding, str(port), redirect, ipaddr]
- args[0] = Utilities.getExecutablePath(args[0])
- process = self.__startProcess(args[0], args[1:])
- if process is None:
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>The debugger backend could not be"""
- """ started.</p>"""))
-
- # set translation function
- if Preferences.getDebugger("PathTranslation"):
- self.translateRemote = \
- Preferences.getDebugger("PathTranslationRemote")
- self.translateLocal = \
- Preferences.getDebugger("PathTranslationLocal")
- self.translate = self.__remoteTranslation
- else:
- self.translate = self.__identityTranslation
- return process, self.__isNetworked, ""
-
- # set translation function
- self.translate = self.__identityTranslation
-
- # setup the environment for the debugger
- if Preferences.getDebugger("DebugEnvironmentReplace"):
- clientEnv = {}
- else:
- clientEnv = os.environ.copy()
- envlist = Utilities.parseEnvironmentString(
- Preferences.getDebugger("DebugEnvironment"))
- for el in envlist:
- try:
- key, value = el.split('=', 1)
- if value.startswith('"') or value.startswith("'"):
- value = value[1:-1]
- clientEnv[str(key)] = str(value)
- except ValueError:
- pass
-
- ipaddr = self.debugServer.getHostAddress(True)
- if runInConsole or Preferences.getDebugger("ConsoleDbgEnabled"):
- ccmd = Preferences.getDebugger("ConsoleDbgCommand")
- if ccmd:
- args = Utilities.parseOptionString(ccmd) + \
- [interpreter, os.path.abspath(debugClient),
- noencoding, str(port), '0', ipaddr]
- args[0] = Utilities.getExecutablePath(args[0])
- process = self.__startProcess(args[0], args[1:], clientEnv)
- if process is None:
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>The debugger backend could not be"""
- """ started.</p>"""))
- return process, self.__isNetworked, interpreter
-
- process = self.__startProcess(
- interpreter,
- [debugClient, noencoding, str(port), redirect, ipaddr],
- clientEnv)
- if process is None:
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>The debugger backend could not be started.</p>"""))
- return process, self.__isNetworked, interpreter
-
- def startRemoteForProject(self, port, runInConsole):
- """
- Public method to start a remote Python interpreter for a project.
-
- @param port portnumber the debug server is listening on (integer)
- @param runInConsole flag indicating to start the debugger in a
- console window (boolean)
- @return client process object (QProcess), a flag to indicate
- a network connection (boolean) and the name of the interpreter
- in case of a local execution (string)
- """
- project = e5App().getObject("Project")
- if not project.isDebugPropertiesLoaded():
- return None, self.__isNetworked, ""
-
- # start debugger with project specific settings
- interpreter = project.getDebugProperty("INTERPRETER")
- debugClient = project.getDebugProperty("DEBUGCLIENT")
-
- redirect = str(project.getDebugProperty("REDIRECT"))
- noencoding = \
- project.getDebugProperty("NOENCODING") and '--no-encoding' or ''
-
- if project.getDebugProperty("REMOTEDEBUGGER"):
- ipaddr = self.debugServer.getHostAddress(False)
- rexec = project.getDebugProperty("REMOTECOMMAND")
- rhost = project.getDebugProperty("REMOTEHOST")
- if rhost == "":
- rhost = "localhost"
- if rexec:
- args = Utilities.parseOptionString(rexec) + \
- [rhost, interpreter, os.path.abspath(debugClient),
- noencoding, str(port), redirect, ipaddr]
- args[0] = Utilities.getExecutablePath(args[0])
- process = self.__startProcess(args[0], args[1:])
- if process is None:
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>The debugger backend could not be"""
- """ started.</p>"""))
- # set translation function
- if project.getDebugProperty("PATHTRANSLATION"):
- self.translateRemote = \
- project.getDebugProperty("REMOTEPATH")
- self.translateLocal = \
- project.getDebugProperty("LOCALPATH")
- self.translate = self.__remoteTranslation
- else:
- self.translate = self.__identityTranslation
- return process, self.__isNetworked, ""
-
- # set translation function
- self.translate = self.__identityTranslation
-
- # setup the environment for the debugger
- if project.getDebugProperty("ENVIRONMENTOVERRIDE"):
- clientEnv = {}
- else:
- clientEnv = os.environ.copy()
- envlist = Utilities.parseEnvironmentString(
- project.getDebugProperty("ENVIRONMENTSTRING"))
- for el in envlist:
- try:
- key, value = el.split('=', 1)
- if value.startswith('"') or value.startswith("'"):
- value = value[1:-1]
- clientEnv[str(key)] = str(value)
- except ValueError:
- pass
-
- ipaddr = self.debugServer.getHostAddress(True)
- if runInConsole or project.getDebugProperty("CONSOLEDEBUGGER"):
- ccmd = project.getDebugProperty("CONSOLECOMMAND") or \
- Preferences.getDebugger("ConsoleDbgCommand")
- if ccmd:
- args = Utilities.parseOptionString(ccmd) + \
- [interpreter, os.path.abspath(debugClient),
- noencoding, str(port), '0', ipaddr]
- args[0] = Utilities.getExecutablePath(args[0])
- process = self.__startProcess(args[0], args[1:], clientEnv)
- if process is None:
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>The debugger backend could not be"""
- """ started.</p>"""))
- return process, self.__isNetworked, interpreter
-
- process = self.__startProcess(
- interpreter,
- [debugClient, noencoding, str(port), redirect, ipaddr],
- clientEnv)
- if process is None:
- E5MessageBox.critical(
- None,
- self.tr("Start Debugger"),
- self.tr(
- """<p>The debugger backend could not be started.</p>"""))
- return process, self.__isNetworked, interpreter
-
- def getClientCapabilities(self):
- """
- Public method to retrieve the debug clients capabilities.
-
- @return debug client capabilities (integer)
- """
- return self.clientCapabilities
-
- def newConnection(self, sock):
- """
- Public slot to handle a new connection.
-
- @param sock reference to the socket object (QTcpSocket)
- @return flag indicating success (boolean)
- """
- # If we already have a connection, refuse this one. It will be closed
- # automatically.
- if self.qsock is not None:
- return False
-
- sock.disconnected.connect(self.debugServer.startClient)
- sock.readyRead.connect(self.__parseClientLine)
-
- self.qsock = sock
-
- # Get the remote clients capabilities
- self.remoteCapabilities()
- return True
-
- def flush(self):
- """
- Public slot to flush the queue.
- """
- # Send commands that were waiting for the connection.
- for cmd in self.queue:
- self.qsock.write(cmd.encode('utf8'))
-
- self.queue = []
-
- def shutdown(self):
- """
- Public method to cleanly shut down.
-
- It closes our socket and shuts down
- the debug client. (Needed on Win OS)
- """
- if self.qsock is None:
- return
-
- # do not want any slots called during shutdown
- self.qsock.disconnected.disconnect(self.debugServer.startClient)
- self.qsock.readyRead.disconnect(self.__parseClientLine)
-
- # close down socket, and shut down client as well.
- self.__sendCommand('{0}\n'.format(DebugProtocol.RequestShutdown))
- self.qsock.flush()
-
- self.qsock.close()
-
- # reinitialize
- self.qsock = None
- self.queue = []
-
- def isConnected(self):
- """
- Public method to test, if a debug client has connected.
-
- @return flag indicating the connection status (boolean)
- """
- return self.qsock is not None
-
- def remoteEnvironment(self, env):
- """
- Public method to set the environment for a program to debug, run, ...
-
- @param env environment settings (dictionary)
- """
- self.__sendCommand('{0}{1}\n'.format(
- DebugProtocol.RequestEnv, str(env)))
-
- def remoteLoad(self, fn, argv, wd, traceInterpreter=False,
- autoContinue=True, autoFork=False, forkChild=False):
- """
- Public method to load a new program to debug.
-
- @param fn the filename to debug (string)
- @param argv the commandline arguments to pass to the program (string)
- @param wd the working directory for the program (string)
- @keyparam traceInterpreter flag indicating if the interpreter library
- should be traced as well (boolean)
- @keyparam autoContinue flag indicating, that the debugger should not
- stop at the first executable line (boolean)
- @keyparam autoFork flag indicating the automatic fork mode (boolean)
- @keyparam forkChild flag indicating to debug the child after forking
- (boolean)
- """
- self.__autoContinue = autoContinue
- self.__scriptName = os.path.abspath(fn)
-
- wd = self.translate(wd, False)
- fn = self.translate(os.path.abspath(fn), False)
- self.__sendCommand('{0}{1}\n'.format(
- DebugProtocol.RequestForkMode, repr((autoFork, forkChild))))
- self.__sendCommand('{0}{1}|{2}|{3}|{4:d}\n'.format(
- DebugProtocol.RequestLoad, wd, fn,
- str(Utilities.parseOptionString(argv)),
- traceInterpreter))
-
- def remoteRun(self, fn, argv, wd, autoFork=False, forkChild=False):
- """
- Public method to load a new program to run.
-
- @param fn the filename to run (string)
- @param argv the commandline arguments to pass to the program (string)
- @param wd the working directory for the program (string)
- @keyparam autoFork flag indicating the automatic fork mode (boolean)
- @keyparam forkChild flag indicating to debug the child after forking
- (boolean)
- """
- self.__scriptName = os.path.abspath(fn)
-
- wd = self.translate(wd, False)
- fn = self.translate(os.path.abspath(fn), False)
- self.__sendCommand('{0}{1}\n'.format(
- DebugProtocol.RequestForkMode, repr((autoFork, forkChild))))
- self.__sendCommand('{0}{1}|{2}|{3}\n'.format(
- DebugProtocol.RequestRun, wd, fn,
- str(Utilities.parseOptionString(argv))))
-
- def remoteCoverage(self, fn, argv, wd, erase=False):
- """
- Public method to load a new program to collect coverage data.
-
- @param fn the filename to run (string)
- @param argv the commandline arguments to pass to the program (string)
- @param wd the working directory for the program (string)
- @keyparam erase flag indicating that coverage info should be
- cleared first (boolean)
- """
- self.__scriptName = os.path.abspath(fn)
-
- wd = self.translate(wd, False)
- fn = self.translate(os.path.abspath(fn), False)
- self.__sendCommand('{0}{1}@@{2}@@{3}@@{4:d}\n'.format(
- DebugProtocol.RequestCoverage, wd, fn,
- str(Utilities.parseOptionString(argv)), erase))
-
- def remoteProfile(self, fn, argv, wd, erase=False):
- """
- Public method to load a new program to collect profiling data.
-
- @param fn the filename to run (string)
- @param argv the commandline arguments to pass to the program (string)
- @param wd the working directory for the program (string)
- @keyparam erase flag indicating that timing info should be cleared
- first (boolean)
- """
- self.__scriptName = os.path.abspath(fn)
-
- wd = self.translate(wd, False)
- fn = self.translate(os.path.abspath(fn), False)
- self.__sendCommand('{0}{1}|{2}|{3}|{4:d}\n'.format(
- DebugProtocol.RequestProfile, wd, fn,
- str(Utilities.parseOptionString(argv)), erase))
-
- def remoteStatement(self, stmt):
- """
- Public method to execute a Python statement.
-
- @param stmt the Python statement to execute (string). It
- should not have a trailing newline.
- """
- self.__sendCommand('{0}\n'.format(stmt))
- self.__sendCommand(DebugProtocol.RequestOK + '\n')
-
- def remoteStep(self):
- """
- Public method to single step the debugged program.
- """
- self.__sendCommand(DebugProtocol.RequestStep + '\n')
-
- def remoteStepOver(self):
- """
- Public method to step over the debugged program.
- """
- self.__sendCommand(DebugProtocol.RequestStepOver + '\n')
-
- def remoteStepOut(self):
- """
- Public method to step out the debugged program.
- """
- self.__sendCommand(DebugProtocol.RequestStepOut + '\n')
-
- def remoteStepQuit(self):
- """
- Public method to stop the debugged program.
- """
- self.__sendCommand(DebugProtocol.RequestStepQuit + '\n')
-
- def remoteContinue(self, special=False):
- """
- Public method to continue the debugged program.
-
- @param special flag indicating a special continue operation (boolean)
- """
- self.__sendCommand('{0}{1:d}\n'.format(
- DebugProtocol.RequestContinue, special))
-
- def remoteBreakpoint(self, fn, line, set, cond=None, temp=False):
- """
- Public method to set or clear a breakpoint.
-
- @param fn filename the breakpoint belongs to (string)
- @param line linenumber of the breakpoint (int)
- @param set flag indicating setting or resetting a breakpoint (boolean)
- @param cond condition of the breakpoint (string)
- @param temp flag indicating a temporary breakpoint (boolean)
- """
- fn = self.translate(fn, False)
- self.__sendCommand('{0}{1}@@{2:d}@@{3:d}@@{4:d}@@{5}\n'.format(
- DebugProtocol.RequestBreak, fn, line, temp, set,
- cond))
-
- def remoteBreakpointEnable(self, fn, line, enable):
- """
- Public method to enable or disable a breakpoint.
-
- @param fn filename the breakpoint belongs to (string)
- @param line linenumber of the breakpoint (int)
- @param enable flag indicating enabling or disabling a breakpoint
- (boolean)
- """
- fn = self.translate(fn, False)
- self.__sendCommand('{0}{1},{2:d},{3:d}\n'.format(
- DebugProtocol.RequestBreakEnable, fn, line, enable))
-
- def remoteBreakpointIgnore(self, fn, line, count):
- """
- Public method to ignore a breakpoint the next couple of occurrences.
-
- @param fn filename the breakpoint belongs to (string)
- @param line linenumber of the breakpoint (int)
- @param count number of occurrences to ignore (int)
- """
- fn = self.translate(fn, False)
- self.__sendCommand('{0}{1},{2:d},{3:d}\n'.format(
- DebugProtocol.RequestBreakIgnore, fn, line, count))
-
- def remoteWatchpoint(self, cond, set, temp=False):
- """
- Public method to set or clear a watch expression.
-
- @param cond expression of the watch expression (string)
- @param set flag indicating setting or resetting a watch expression
- (boolean)
- @param temp flag indicating a temporary watch expression (boolean)
- """
- # cond is combination of cond and special (s. watch expression viewer)
- self.__sendCommand('{0}{1}@@{2:d}@@{3:d}\n'.format(
- DebugProtocol.RequestWatch, cond, temp, set))
-
- def remoteWatchpointEnable(self, cond, enable):
- """
- Public method to enable or disable a watch expression.
-
- @param cond expression of the watch expression (string)
- @param enable flag indicating enabling or disabling a watch expression
- (boolean)
- """
- # cond is combination of cond and special (s. watch expression viewer)
- self.__sendCommand('{0}{1},{2:d}\n'.format(
- DebugProtocol.RequestWatchEnable, cond, enable))
-
- def remoteWatchpointIgnore(self, cond, count):
- """
- Public method to ignore a watch expression the next couple of
- occurrences.
-
- @param cond expression of the watch expression (string)
- @param count number of occurrences to ignore (int)
- """
- # cond is combination of cond and special (s. watch expression viewer)
- self.__sendCommand('{0}{1},{2:d}\n'.format(
- DebugProtocol.RequestWatchIgnore, cond, count))
-
- def remoteRawInput(self, s):
- """
- Public method to send the raw input to the debugged program.
-
- @param s the raw input (string)
- """
- self.__sendCommand(s + '\n')
-
- def remoteThreadList(self):
- """
- Public method to request the list of threads from the client.
- """
- self.__sendCommand('{0}\n'.format(DebugProtocol.RequestThreadList))
-
- def remoteSetThread(self, tid):
- """
- Public method to request to set the given thread as current thread.
-
- @param tid id of the thread (integer)
- """
- self.__sendCommand('{0}{1:d}\n'.format(
- DebugProtocol.RequestThreadSet, tid))
-
- def remoteClientVariables(self, scope, filter, framenr=0):
- """
- Public method to request the variables of the debugged program.
-
- @param scope the scope of the variables (0 = local, 1 = global)
- @param filter list of variable types to filter out (list of int)
- @param framenr framenumber of the variables to retrieve (int)
- """
- self.__sendCommand('{0}{1:d}, {2:d}, {3}\n'.format(
- DebugProtocol.RequestVariables, framenr, scope, str(filter)))
-
- def remoteClientVariable(self, scope, filter, var, framenr=0):
- """
- Public method to request the variables of the debugged program.
-
- @param scope the scope of the variables (0 = local, 1 = global)
- @param filter list of variable types to filter out (list of int)
- @param var list encoded name of variable to retrieve (string)
- @param framenr framenumber of the variables to retrieve (int)
- """
- self.__sendCommand('{0}{1}, {2:d}, {3:d}, {4}\n'.format(
- DebugProtocol.RequestVariable, str(var), framenr, scope,
- str(filter)))
-
- def remoteClientSetFilter(self, scope, filter):
- """
- Public method to set a variables filter list.
-
- @param scope the scope of the variables (0 = local, 1 = global)
- @param filter regexp string for variable names to filter out (string)
- """
- self.__sendCommand('{0}{1:d}, "{2}"\n'.format(
- DebugProtocol.RequestSetFilter, scope, filter))
-
- def setCallTraceEnabled(self, on):
- """
- Public method to set the call trace state.
-
- @param on flag indicating to enable the call trace function (boolean)
- """
- if on:
- cmd = "on"
- else:
- cmd = "off"
- self.__sendCommand('{0}{1}\n'.format(
- DebugProtocol.RequestCallTrace, cmd))
-
- def remoteEval(self, arg):
- """
- Public method to evaluate arg in the current context of the debugged
- program.
-
- @param arg the arguments to evaluate (string)
- """
- self.__sendCommand('{0}{1}\n'.format(DebugProtocol.RequestEval, arg))
-
- def remoteExec(self, stmt):
- """
- Public method to execute stmt in the current context of the debugged
- program.
-
- @param stmt statement to execute (string)
- """
- self.__sendCommand('{0}{1}\n'.format(DebugProtocol.RequestExec, stmt))
-
- def remoteBanner(self):
- """
- Public slot to get the banner info of the remote client.
- """
- self.__sendCommand(DebugProtocol.RequestBanner + '\n')
-
- def remoteCapabilities(self):
- """
- Public slot to get the debug clients capabilities.
- """
- self.__sendCommand(DebugProtocol.RequestCapabilities + '\n')
-
- def remoteCompletion(self, text):
- """
- Public slot to get the a list of possible commandline completions
- from the remote client.
-
- @param text the text to be completed (string)
- """
- self.__sendCommand("{0}{1}\n".format(
- DebugProtocol.RequestCompletion, text))
-
- def remoteUTPrepare(self, fn, tn, tfn, failed, cov, covname, coverase):
- """
- Public method to prepare a new unittest run.
-
- @param fn the filename to load (string)
- @param tn the testname to load (string)
- @param tfn the test function name to load tests from (string)
- @param failed list of failed test, if only failed test should be run
- (list of strings)
- @param cov flag indicating collection of coverage data is requested
- (boolean)
- @param covname filename to be used to assemble the coverage caches
- filename (string)
- @param coverase flag indicating erasure of coverage data is requested
- (boolean)
- """
- self.__scriptName = os.path.abspath(fn)
-
- fn = self.translate(os.path.abspath(fn), False)
- self.__sendCommand('{0}{1}|{2}|{3}|{4}|{5:d}|{6}|{7:d}\n'.format(
- DebugProtocol.RequestUTPrepare, fn, tn, tfn, str(failed),
- cov, covname, coverase))
-
- def remoteUTRun(self):
- """
- Public method to start a unittest run.
- """
- self.__sendCommand('{0}\n'.format(DebugProtocol.RequestUTRun))
-
- def remoteUTStop(self):
- """
- Public method to stop a unittest run.
- """
- self.__sendCommand('{0}\n'.format(DebugProtocol.RequestUTStop))
-
- def __askForkTo(self):
- """
- Private method to ask the user which branch of a fork to follow.
- """
- selections = [self.tr("Parent Process"),
- self.tr("Child process")]
- res, ok = QInputDialog.getItem(
- None,
- self.tr("Client forking"),
- self.tr("Select the fork branch to follow."),
- selections,
- 0, False)
- if not ok or res == selections[0]:
- self.__sendCommand(DebugProtocol.ResponseForkTo + 'parent\n')
- else:
- self.__sendCommand(DebugProtocol.ResponseForkTo + 'child\n')
-
- def __parseClientLine(self):
- """
- Private method to handle data from the client.
- """
- while self.qsock and self.qsock.canReadLine():
- qs = self.qsock.readLine()
- if self.codec is not None:
- line = self.codec.toUnicode(qs)
- else:
- line = bytes(qs).decode()
- if line.endswith(DebugProtocol.EOT):
- line = line[:-len(DebugProtocol.EOT)]
- if not line:
- continue
-
-## print("Server: ", line) ##debug
-
- eoc = line.find('<') + 1
-
- # Deal with case where user has written directly to stdout
- # or stderr, but not line terminated and we stepped over the
- # write call, in that case the >line< will not be the first
- # string read from the socket...
- boc = line.find('>')
- if boc > 0 and eoc > boc:
- self.debugServer.signalClientOutput(line[:boc])
- line = line[boc:]
- eoc = line.find('<') + 1
- boc = line.find('>')
-
- if boc >= 0 and eoc > boc:
- resp = line[boc:eoc]
- evalArg = self.__unicodeRe.sub(r"\1", line[eoc:-1])
-
- if resp == DebugProtocol.ResponseLine or \
- resp == DebugProtocol.ResponseStack:
- stack = eval(evalArg)
- for s in stack:
- s[0] = self.translate(s[0], True)
- cf = stack[0]
- if self.__autoContinue:
- self.__autoContinue = False
- QTimer.singleShot(0, self.remoteContinue)
- else:
- self.debugServer.signalClientLine(
- cf[0], int(cf[1]),
- resp == DebugProtocol.ResponseStack)
- self.debugServer.signalClientStack(stack)
- continue
-
- if resp == DebugProtocol.CallTrace:
- event, fromStr, toStr = line[eoc:-1].split("@@")
- isCall = event.lower() == "c"
- fromFile, fromLineno, fromFunc = fromStr.rsplit(":", 2)
- toFile, toLineno, toFunc = toStr.rsplit(":", 2)
- self.debugServer.signalClientCallTrace(isCall, fromFile,
- fromLineno,
- fromFunc,
- toFile, toLineno,
- toFunc)
- continue
-
- if resp == DebugProtocol.ResponseThreadList:
- currentId, threadList = eval(evalArg)
- self.debugServer.signalClientThreadList(
- currentId, threadList)
- continue
-
- if resp == DebugProtocol.ResponseThreadSet:
- self.debugServer.signalClientThreadSet()
- continue
-
- if resp == DebugProtocol.ResponseVariables:
- vlist = eval(evalArg)
- scope = vlist[0]
- try:
- variables = vlist[1:]
- except IndexError:
- variables = []
- self.debugServer.signalClientVariables(scope, variables)
- continue
-
- if resp == DebugProtocol.ResponseVariable:
- vlist = eval(evalArg)
- scope = vlist[0]
- try:
- variables = vlist[1:]
- except IndexError:
- variables = []
- self.debugServer.signalClientVariable(scope, variables)
- continue
-
- if resp == DebugProtocol.ResponseOK:
- self.debugServer.signalClientStatement(False)
- continue
-
- if resp == DebugProtocol.ResponseContinue:
- self.debugServer.signalClientStatement(True)
- continue
-
- if resp == DebugProtocol.ResponseException:
- exc = self.translate(evalArg, True)
- try:
- exclist = eval(exc)
- exctype = exclist[0]
- excmessage = exclist[1]
- stack = exclist[2:]
- if stack and stack[0] and stack[0][0] == "<string>":
- for stackEntry in stack:
- if stackEntry[0] == "<string>":
- stackEntry[0] = self.__scriptName
- else:
- break
- except (IndexError, ValueError, SyntaxError):
- exctype = None
- excmessage = ''
- stack = []
- self.debugServer.signalClientException(
- exctype, excmessage, stack)
- continue
-
- if resp == DebugProtocol.ResponseSyntax:
- exc = self.translate(evalArg, True)
- try:
- message, (fn, ln, cn) = eval(exc)
- if fn is None:
- fn = ''
- except (IndexError, ValueError):
- message = None
- fn = ''
- ln = 0
- cn = 0
- if cn is None:
- cn = 0
- self.debugServer.signalClientSyntaxError(
- message, fn, ln, cn)
- continue
-
- if resp == DebugProtocol.ResponseSignal:
- sig = self.translate(evalArg, True)
- message, (fn, ln, func, args) = eval(sig)
- self.debugServer.signalClientSignal(
- message, fn, ln, func, args)
- continue
-
- if resp == DebugProtocol.ResponseExit:
- self.__scriptName = ""
- self.debugServer.signalClientExit(evalArg)
- continue
-
- if resp == DebugProtocol.ResponseClearBreak:
- fn, lineno = evalArg.split(',')
- lineno = int(lineno)
- fn = self.translate(fn, True)
- self.debugServer.signalClientClearBreak(fn, lineno)
- continue
-
- if resp == DebugProtocol.ResponseBPConditionError:
- fn, lineno = evalArg.split(',')
- lineno = int(lineno)
- fn = self.translate(fn, True)
- self.debugServer.signalClientBreakConditionError(
- fn, lineno)
- continue
-
- if resp == DebugProtocol.ResponseClearWatch:
- self.debugServer.signalClientClearWatch(evalArg)
- continue
-
- if resp == DebugProtocol.ResponseWPConditionError:
- self.debugServer.signalClientWatchConditionError(evalArg)
- continue
-
- if resp == DebugProtocol.ResponseRaw:
- prompt, echo = eval(evalArg)
- self.debugServer.signalClientRawInput(prompt, echo)
- continue
-
- if resp == DebugProtocol.ResponseBanner:
- version, platform, dbgclient = eval(evalArg)
- self.debugServer.signalClientBanner(
- version, platform, dbgclient)
- continue
-
- if resp == DebugProtocol.ResponseCapabilities:
- cap, clType = eval(evalArg)
- self.clientCapabilities = cap
- self.debugServer.signalClientCapabilities(cap, clType)
- continue
-
- if resp == DebugProtocol.ResponseCompletion:
- clstring, text = evalArg.split('||')
- cl = eval(clstring)
- self.debugServer.signalClientCompletionList(cl, text)
- continue
-
- if resp == DebugProtocol.PassiveStartup:
- fn, exc = evalArg.split('|')
- exc = bool(exc)
- fn = self.translate(fn, True)
- self.debugServer.passiveStartUp(fn, exc)
- continue
-
- if resp == DebugProtocol.ResponseUTPrepared:
- res, exc_type, exc_value = eval(evalArg)
- self.debugServer.clientUtPrepared(res, exc_type, exc_value)
- continue
-
- if resp == DebugProtocol.ResponseUTStartTest:
- testname, doc = eval(evalArg)
- self.debugServer.clientUtStartTest(testname, doc)
- continue
-
- if resp == DebugProtocol.ResponseUTStopTest:
- self.debugServer.clientUtStopTest()
- continue
-
- if resp == DebugProtocol.ResponseUTTestFailed:
- testname, traceback, id = eval(evalArg)
- self.debugServer.clientUtTestFailed(
- testname, traceback, id)
- continue
-
- if resp == DebugProtocol.ResponseUTTestErrored:
- testname, traceback, id = eval(evalArg)
- self.debugServer.clientUtTestErrored(
- testname, traceback, id)
- continue
-
- if resp == DebugProtocol.ResponseUTTestSkipped:
- testname, reason, id = eval(line[eoc:-1])
- self.debugServer.clientUtTestSkipped(testname, reason, id)
- continue
-
- if resp == DebugProtocol.ResponseUTTestFailedExpected:
- testname, traceback, id = eval(line[eoc:-1])
- self.debugServer.clientUtTestFailedExpected(
- testname, traceback, id)
- continue
-
- if resp == DebugProtocol.ResponseUTTestSucceededUnexpected:
- testname, id = eval(line[eoc:-1])
- self.debugServer.clientUtTestSucceededUnexpected(
- testname, id)
- continue
-
- if resp == DebugProtocol.ResponseUTFinished:
- self.debugServer.clientUtFinished()
- continue
-
- if resp == DebugProtocol.RequestForkTo:
- self.__askForkTo()
- continue
-
- self.debugServer.signalClientOutput(line)
-
- def __sendCommand(self, cmd):
- """
- Private method to send a single line command to the client.
-
- @param cmd command to send to the debug client (string)
- """
- if self.qsock is not None:
- self.qsock.write(cmd.encode('utf8'))
- else:
- self.queue.append(cmd)
-
-
-def createDebuggerInterfacePython(debugServer, passive):
- """
- Module function to create a debugger interface instance.
-
-
- @param debugServer reference to the debug server
- @type DebugServer
- @param passive flag indicating passive connection mode
- @type bool
- @return instantiated debugger interface
- @rtype DebuggerInterfacePython
- """
- return DebuggerInterfacePython(debugServer, passive)
-
-
-def getRegistryData():
- """
- Module function to get characterizing data for the debugger interface.
-
- @return tuple containing client type, client capabilities, client file
- type associations and reference to creation function
- @rtype tuple of (str, int, list of str, function)
- """
- exts = []
- for ext in Preferences.getDebugger("PythonExtensions").split():
- if ext.startswith("."):
- exts.append(ext)
- else:
- exts.append(".{0}".format(ext))
-
- if exts and Preferences.getDebugger("PythonInterpreter"):
- return ["Python2", ClientDefaultCapabilities, exts,
- createDebuggerInterfacePython]
- else:
- return ["", 0, [], None]
