eric: comparison DebugClients/Python/DebugClientBase.py

-:14680839ad7a
+:d062df8f1d9f
 import json
 import imp
 import re
 import atexit
 import signal
+import time
 import DebugClientCapabilities
 import DebugVariables
 from DebugBase import setRecursionLimit, printerr   # __IGNORE_WARNING__
 self.sessionClose()
 elif method == "RequestCompletion":
 self.__completionList(params["text"])
-elif method == "RequestUTPrepare":
+elif method == "RequestUTDiscover":
-sys.path.insert(
+if params["syspath"]:
-0, os.path.dirname(os.path.abspath(params["filename"])))
+sys.path = params["syspath"] + sys.path
-os.chdir(sys.path[0])
+discoveryStart = params["discoverystart"]
+if not discoveryStart:
+discoveryStart = params["workdir"]
+os.chdir(params["discoverystart"])
 # 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(
+testLoader = unittest.TestLoader()
-params["testname"], params["filename"])
+test = testLoader.discover(discoveryStart)
-try:
+if hasattr(testLoader, "errors") and \
+bool(testLoader.errors):
+self.sendJsonCommand("ResponseUTDiscover", {
+"testCasesList": [],
+"exception": "DiscoveryError",
+"message": "\n\n".join(testLoader.errors),
+})
+else:
+testsList = self.__assembleTestCasesList(test,
+discoveryStart)
+self.sendJsonCommand("ResponseUTDiscover", {
+"testCasesList": testsList,
+"exception": "",
+"message": "",
+})
+except Exception:
+exc_type, exc_value, exc_tb = sys.exc_info()
+self.sendJsonCommand("ResponseUTDiscover", {
+"testCasesList": [],
+"exception": exc_type.__name__,
+"message": str(exc_value),
+})
+elif method == "RequestUTPrepare":
+if params["syspath"]:
+sys.path = params["syspath"] + sys.path
+sys.path.insert(
+0, os.path.dirname(os.path.abspath(params["filename"])))
+if params["workdir"]:
+os.chdir(params["workdir"])
+else:
+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
+testLoader = unittest.TestLoader()
+if params["discover"]:
+discoveryStart = params["discoverystart"]
+if not discoveryStart:
+discoveryStart = params["workdir"]
+if params["testcases"]:
+self.test = testLoader.loadTestsFromNames(
+params["testcases"])
+else:
+self.test = testLoader.discover(discoveryStart)
+else:
+if params["filename"]:
+utModule = imp.load_source(
+params["testname"], params["filename"])
+else:
+utModule = None
 if params["failed"]:
-self.test = unittest.defaultTestLoader\
+if utModule:
-.loadTestsFromNames(params["failed"], utModule)
+failed = [t.split(".", 1)[1]
+for t in params["failed"]]
+else:
+failed = params["failed"][:]
+self.test = testLoader.loadTestsFromNames(
+failed, utModule)
 else:
-self.test = unittest.defaultTestLoader\
+self.test = testLoader.loadTestsFromName(
-.loadTestsFromName(params["testfunctionname"],
+params["testfunctionname"], utModule)
-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__,
 if params["coverageerase"]:
 self.cover.erase()
 else:
 self.cover = None
+if params["debug"]:
+Breakpoint.clear_all_breaks()
+Watch.clear_all_watches()
 self.sendJsonCommand("ResponseUTPrepared", {
 "count": self.test.countTestCases(),
 "exception": "",
 "message": "",
 })
 elif method == "RequestUTRun":
 from DCTestResult import DCTestResult
-self.testResult = DCTestResult(self)
+self.testResult = DCTestResult(self, params["failfast"])
 if self.cover:
 self.cover.start()
-self.test.run(self.testResult)
+self.debugging = params["debug"]
+if params["debug"]:
+locals_ = locals()
+self.threads.clear()
+self.attachThread(mainThread=True)
+sys.setprofile(None)
+self.mainThread.run(
+"result = self.test.run(self.testResult)\n",
+localsDict=locals_)
+result = locals_["result"]
+else:
+result = self.test.run(self.testResult)
 if self.cover:
 self.cover.stop()
 self.cover.save()
-self.sendJsonCommand("ResponseUTFinished", {})
+self.sendJsonCommand("ResponseUTFinished", {
+"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.
+@param suite test suite to be inspected
+@type unittest.TestSuite
+@param start name of directory discovery was started at
+@type str
+@return list of tuples containing the test case ID, a short description
+and the path of the test file name
+@rtype list of tuples of (str, str, str)
+"""
+import unittest
+testCases = []
+for test in suite:
+if isinstance(test, unittest.TestSuite):
+testCases.extend(self.__assembleTestCasesList(test, start))
+else:
+testId = test.id()
+if "ModuleImportFailure" not in testId and \
+"LoadTestsFailure" not in testId and \
+"_FailedTest" not in testId:
+filename = os.path.join(
+start,
+test.__module__.replace(".", os.sep) + ".py")
+testCases.append(
+(test.id(), test.shortDescription(), filename)
+)
+return testCases
 def sendJsonCommand(self, method, params):
 """
 Public method to send a single command or response to the IDE.
 version = 'v4'
 if version == 'v4':
 family = socket.AF_INET
 else:
 family = socket.AF_INET6
-return socket.getaddrinfo(host, None, family,
-socket.SOCK_STREAM)[0][4][0]
+retryCount = 0
+while retryCount < 10:
+try:
+addrinfo = socket.getaddrinfo(
+host, None, family, socket.SOCK_STREAM)
+return addrinfo[0][4][0]
+except Exception:
+retryCount += 1
+time.sleep(3)
+return None
 def main(self):
 """
 Public method implementing the main method.
 """
 try:
 redirect = int(sys.argv[2])
 except (ValueError, IndexError):
 redirect = True
-try:
+ipOrHost = sys.argv[3]
-ipOrHost = sys.argv[3]
+if ':' in ipOrHost:
-if ':' in ipOrHost:
+# IPv6 address
 remoteAddress = ipOrHost
 elif ipOrHost[0] in '0123456789':
-remoteAddress = ipOrHost
+# IPv4 address
-else:
+remoteAddress = ipOrHost
-remoteAddress = self.__resolveHost(ipOrHost)
+else:
-except Exception:
+remoteAddress = self.__resolveHost(ipOrHost)
-remoteAddress = None
 sys.argv = ['']
 if '' not in sys.path:
 sys.path.insert(0, '')

Mercurial Repositories > eric / file comparison

comparison: DebugClients/Python/DebugClientBase.py

DebugClients/Python/DebugClientBase.py