797 elif method == "RequestSetNoDebugList": |
797 elif method == "RequestSetNoDebugList": |
798 self.noDebugList = params["noDebug"][:] |
798 self.noDebugList = params["noDebug"][:] |
799 |
799 |
800 elif method == "RequestCompletion": |
800 elif method == "RequestCompletion": |
801 self.__completionList(params["text"]) |
801 self.__completionList(params["text"]) |
802 |
|
803 elif method == "RequestUTDiscover": |
|
804 if params["syspath"]: |
|
805 sys.path = params["syspath"] + sys.path |
|
806 |
|
807 discoveryStart = params["discoverystart"] |
|
808 if not discoveryStart: |
|
809 discoveryStart = params["workdir"] |
|
810 |
|
811 top_level_dir = params["workdir"] |
|
812 |
|
813 os.chdir(params["discoverystart"]) |
|
814 |
|
815 # set the system exception handling function to ensure, that |
|
816 # we report on all unhandled exceptions |
|
817 sys.excepthook = self.__unhandled_exception |
|
818 self.__interceptSignals() |
|
819 |
|
820 try: |
|
821 import unittest |
|
822 testLoader = unittest.TestLoader() |
|
823 test = testLoader.discover( |
|
824 discoveryStart, top_level_dir=top_level_dir) |
|
825 if (hasattr(testLoader, "errors") and |
|
826 bool(testLoader.errors)): |
|
827 self.sendJsonCommand("ResponseUTDiscover", { |
|
828 "testCasesList": [], |
|
829 "exception": "DiscoveryError", |
|
830 "message": "\n\n".join(testLoader.errors), |
|
831 }) |
|
832 else: |
|
833 testsList = self.__assembleTestCasesList(test, |
|
834 discoveryStart) |
|
835 self.sendJsonCommand("ResponseUTDiscover", { |
|
836 "testCasesList": testsList, |
|
837 "exception": "", |
|
838 "message": "", |
|
839 }) |
|
840 except Exception: |
|
841 exc_type, exc_value, exc_tb = sys.exc_info() |
|
842 self.sendJsonCommand("ResponseUTDiscover", { |
|
843 "testCasesList": [], |
|
844 "exception": exc_type.__name__, |
|
845 "message": str(exc_value), |
|
846 }) |
|
847 |
|
848 elif method == "RequestUTPrepare": |
|
849 if params["syspath"]: |
|
850 sys.path = params["syspath"] + sys.path |
|
851 top_level_dir = None |
|
852 if params["workdir"]: |
|
853 os.chdir(params["workdir"]) |
|
854 top_level_dir = params["workdir"] |
|
855 |
|
856 # set the system exception handling function to ensure, that |
|
857 # we report on all unhandled exceptions |
|
858 sys.excepthook = self.__unhandled_exception |
|
859 self.__interceptSignals() |
|
860 |
|
861 try: |
|
862 import unittest |
|
863 testLoader = unittest.TestLoader() |
|
864 if params["discover"]: |
|
865 discoveryStart = params["discoverystart"] |
|
866 if not discoveryStart: |
|
867 discoveryStart = params["workdir"] |
|
868 sys.path.insert( |
|
869 0, os.path.abspath(discoveryStart)) |
|
870 if params["testcases"]: |
|
871 self.test = testLoader.loadTestsFromNames( |
|
872 params["testcases"]) |
|
873 else: |
|
874 self.test = testLoader.discover( |
|
875 discoveryStart, top_level_dir=top_level_dir) |
|
876 else: |
|
877 sys.path.insert( |
|
878 0, |
|
879 os.path.dirname(os.path.abspath(params["filename"])) |
|
880 ) |
|
881 if params["filename"]: |
|
882 utModule = __import__(params["testname"]) |
|
883 else: |
|
884 utModule = None |
|
885 if params["failed"]: |
|
886 if utModule: |
|
887 failed = [t.split(".", 1)[1] |
|
888 for t in params["failed"]] |
|
889 else: |
|
890 failed = params["failed"][:] |
|
891 self.test = testLoader.loadTestsFromNames( |
|
892 failed, utModule) |
|
893 else: |
|
894 self.test = testLoader.loadTestsFromName( |
|
895 params["testfunctionname"], utModule) |
|
896 except Exception: |
|
897 exc_type, exc_value, exc_tb = sys.exc_info() |
|
898 self.sendJsonCommand("ResponseUTPrepared", { |
|
899 "count": 0, |
|
900 "exception": exc_type.__name__, |
|
901 "message": str(exc_value), |
|
902 }) |
|
903 return |
|
904 |
|
905 # generate a coverage object |
|
906 if params["coverage"]: |
|
907 from coverage import Coverage |
|
908 self.cover = Coverage( |
|
909 auto_data=True, |
|
910 data_file="{0}.coverage".format( |
|
911 os.path.splitext(params["coveragefile"])[0])) |
|
912 if params["coverageerase"]: |
|
913 self.cover.erase() |
|
914 else: |
|
915 self.cover = None |
|
916 |
|
917 if params["debug"]: |
|
918 Breakpoint.clear_all_breaks() |
|
919 Watch.clear_all_watches() |
|
920 |
|
921 self.sendJsonCommand("ResponseUTPrepared", { |
|
922 "count": self.test.countTestCases(), |
|
923 "exception": "", |
|
924 "message": "", |
|
925 }) |
|
926 |
|
927 elif method == "RequestUTRun": |
|
928 from DCTestResult import DCTestResult |
|
929 self.disassembly = None |
|
930 self.testResult = DCTestResult(self, params["failfast"]) |
|
931 if self.cover: |
|
932 self.cover.start() |
|
933 self.debugging = params["debug"] |
|
934 if params["debug"]: |
|
935 self.multiprocessSupport = False |
|
936 locals_ = locals() |
|
937 self.threads.clear() |
|
938 self.attachThread(mainThread=True) |
|
939 sys.setprofile(None) |
|
940 self.running = sys.argv[0] |
|
941 self.mainThread.run( |
|
942 "result = self.test.run(self.testResult)\n", |
|
943 self.debugMod.__dict__, |
|
944 localsDict=locals_, |
|
945 debug=True, |
|
946 closeSession=False) |
|
947 result = locals_["result"] |
|
948 else: |
|
949 result = self.test.run(self.testResult) |
|
950 if self.cover: |
|
951 self.cover.stop() |
|
952 self.cover.save() |
|
953 self.sendJsonCommand("ResponseUTFinished", { |
|
954 "status": 0 if result.wasSuccessful() else 1, |
|
955 }) |
|
956 |
|
957 elif method == "RequestUTStop": |
|
958 self.testResult.stop() |
|
959 |
|
960 def __assembleTestCasesList(self, suite, start): |
|
961 """ |
|
962 Private method to assemble a list of test cases included in a test |
|
963 suite. |
|
964 |
|
965 @param suite test suite to be inspected |
|
966 @type unittest.TestSuite |
|
967 @param start name of directory discovery was started at |
|
968 @type str |
|
969 @return list of tuples containing the test case ID, a short description |
|
970 and the path of the test file name |
|
971 @rtype list of tuples of (str, str, str) |
|
972 """ |
|
973 import unittest |
|
974 testCases = [] |
|
975 for test in suite: |
|
976 if isinstance(test, unittest.TestSuite): |
|
977 testCases.extend(self.__assembleTestCasesList(test, start)) |
|
978 else: |
|
979 testId = test.id() |
|
980 if ("ModuleImportFailure" not in testId and |
|
981 "LoadTestsFailure" not in testId and |
|
982 "_FailedTest" not in testId): |
|
983 filename = os.path.join( |
|
984 start, |
|
985 test.__module__.replace(".", os.sep) + ".py") |
|
986 testCases.append( |
|
987 (test.id(), test.shortDescription(), filename) |
|
988 ) |
|
989 return testCases |
|
990 |
802 |
991 def setDisassembly(self, disassembly): |
803 def setDisassembly(self, disassembly): |
992 """ |
804 """ |
993 Public method to store a disassembly of the code object raising an |
805 Public method to store a disassembly of the code object raising an |
994 exception. |
806 exception. |