eric: comparison eric6/DebugClients/Python/DebugClientBase.py

-:ab8f95bc7d2d
+:eb65864ca038
 from AsyncFile import AsyncFile, AsyncPendingWrite
 from DebugConfig import ConfigQtNames, SpecialAttributes
 from FlexCompleter import Completer
 from DebugUtilities import prepareJsonCommand
 from BreakpointWatch import Breakpoint, Watch
-##from MultiProcessDebugExtension import patchNewProcessFunctions
+from MultiProcessDebugExtension import patchNewProcessFunctions
 from DebugUtilities import getargvalues, formatargvalues
 DebugClientInstance = None
 __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)
 self.running = None
 self.test = None
 self.debugging = False
 self.multiprocessSupport = False
 self.noDebugList = []
-self.fork_auto = False
-self.fork_child = False
 self.readstream = None
 self.writestream = None
 self.errorstream = None
 self.pollingDisabled = False
 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 filename name of the source file
-@param mode kind of code to be generated (string, exec or eval)
+@type str
+@param mode kind of code to be generated (exec or eval)
+@type str
 @return compiled code object (None in case of errors)
 """
 with codecs.open(filename, encoding=self.__coding) as fp:
 statement = fp.read()
+return self.__compileCommand(statement, filename=filename, mode=mode)
+def __compileCommand(self, statement, filename="<string>", mode="exec"):
+"""
+Private method to compile source code.
+@param statement source code string to be compiled
+@type str
+@param filename name of the source file
+@type str
+@param mode kind of code to be generated (exec or eval)
+@type str
+@return compiled code object (None in case of errors)
+"""
 try:
 code = compile(statement + '\n', filename, mode)
 except SyntaxError:
 exctype, excval, exctb = sys.exc_info()
 try:
 self.running = sys.argv[0]
 self.debugging = True
 self.multiprocessSupport = params["multiprocess"]
-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
 else:
 os.chdir(params["workdir"])
 self.running = sys.argv[0]
 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
 "status": 0 if result.wasSuccessful() else 1,
 })
 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 __assembleTestCasesList(self, suite, start):
 """
 Private method to assemble a list of test cases included in a test
 suite.
 self.mainThread.set_trace()
 def startProgInDebugger(self, progargs, wd='', host=None,
 port=None, exceptions=True, tracePython=False,
 redirect=True, passive=True,
-multiprocessSupport=False):
+multiprocessSupport=False, codeStr=""):
 """
 Public method used to start the remote debugger.
 @param progargs commandline for the program to be debugged
 (list of strings)
 @param passive flag indicating a passive debugging session
 @type bool
 @param multiprocessSupport flag indicating to enable multiprocess
 debugging support
 @type bool
+@param codeStr string containing Python code to execute
+@type str
 @return exit code of the debugged program
 @rtype int
 """
 if host is None:
 host = os.getenv('ERICHOST', 'localhost')
 name = os.path.basename(progargs[0])
 self.connectDebugger(port, remoteAddress, redirect, name=name)
 self._fncache = {}
 self.dircache = []
-sys.argv = progargs[:]
+if codeStr:
-sys.argv[0] = os.path.abspath(sys.argv[0])
+self.running = "<string>"
-sys.path = self.__getSysPath(os.path.dirname(sys.argv[0]))
+sys.argv = ["<string>"] + progargs[:]
-if wd == '':
-os.chdir(sys.path[1])
 else:
-os.chdir(wd)
+sys.argv = progargs[:]
-self.running = sys.argv[0]
+sys.argv[0] = os.path.abspath(sys.argv[0])
-self.__setCoding(self.running)
+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.__setCoding(self.running)
 self.debugging = True
 self.multiprocessSupport = multiprocessSupport
 self.passive = passive
 if passive:
 self.__interceptSignals()
 # 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 codeStr:
+code = self.__compileCommand(codeStr)
+else:
+code = self.__compileFileSource(self.running)
 if code:
 res = self.mainThread.run(code, self.debugMod.__dict__, debug=True)
 else:
 res = 42        # should not happen
 return res
 tracePython = False
 exceptions = True
 redirect = True
 passive = True
 multiprocess = False
+hasCode = False
 while args[0]:
 if args[0] == '-h':
 host = args[1]
 del args[0]
 del args[0]
 redirect = False
 del args[0]
 elif args[0] == '--no-encoding':
 self.noencoding = True
 del args[0]
-elif args[0] == '--fork-child':
-self.fork_auto = True
-self.fork_child = True
-del args[0]
-elif args[0] == '--fork-parent':
-self.fork_auto = True
-self.fork_child = False
-del args[0]
 elif args[0] == '--no-passive':
 passive = False
 del args[0]
 elif args[0] == '--multiprocess':
 multiprocess = True
+del args[0]
+elif args[0] == '--code':
+hasCode = True
 del args[0]
 elif args[0] == '--':
 del args[0]
 break
 else:   # unknown option
 else:
 # Store options if a process is spawn
 # TODO: check which ones are really needed
 self.startOptions = (
 wd, host, port, exceptions, tracePython, redirect,
-self.noencoding, self.fork_auto, self.fork_child,
+self.noencoding,
 )
 if not self.noencoding:
 self.__coding = self.defaultCoding
-##                patchNewProcessFunctions(multiprocess, self)
+patchNewProcessFunctions(multiprocess, self)
+if hasCode:
+codeStr = args.pop(0)
+else:
+codeStr=""
 res = self.startProgInDebugger(
 args, wd, host, port, exceptions=exceptions,
 tracePython=tracePython, redirect=redirect,
-passive=passive, multiprocessSupport=multiprocess
+passive=passive, multiprocessSupport=multiprocess,
+codeStr=codeStr
 )
 sys.exit(res)
 else:
 if sys.argv[1] == '--no-encoding':
 self.noencoding = True
 if port >= 0:
 # Store options if a process is spawn
 # TODO: check which ones are really needed
 self.startOptions = (
 '', remoteAddress, port, True, False, redirect,
-self.noencoding, self.fork_auto, self.fork_child,
+self.noencoding,
 )
 if not self.noencoding:
 self.__coding = self.defaultCoding
+patchNewProcessFunctions(self.multiprocessSupport, self)
 self.connectDebugger(port, remoteAddress, redirect)
-##                patchNewProcessFunctions(self.multiprocessSupport, self)
 self.__interact()
 else:
 print("No network port given. Aborting...")
 # __IGNORE_WARNING_M801__
-def fork(self):
-"""
-Public method implementing a fork routine deciding which branch
-to follow.
-@return process ID (integer)
-"""
-##        if not self.fork_auto:
-##            self.sendJsonCommand("RequestForkTo", {})
-##            self.eventLoop(True)
-pid = DebugClientOrigFork()
-if pid == 0:
-# child
-##            if not self.fork_child:
-sys.settrace(None)
-sys.setprofile(None)
-self.sessionClose(False)
-##                (wd, host, port, exceptions, tracePython, redirect,
-##                 noencoding, fork_auto, fork_child) = self.startOptions
-##                self.startDebugger(sys.argv[0], host, port,
-##                                   exceptions=exceptions,
-##                                   tracePython=tracePython,
-##                                   redirect=redirect,
-##                                   passive=False)
-else:
-# parent
-if self.fork_child:
-sys.settrace(None)
-sys.setprofile(None)
-self.sessionClose(False)
-return pid
 def close(self, fd):
 """
 Public method implementing a close method as a replacement for
 os.close().

Mercurial Repositories > eric / file comparison

comparison: eric6/DebugClients/Python/DebugClientBase.py

eric6/DebugClients/Python/DebugClientBase.py