diff -r 22dab1be7953 -r 7f27bf3b50c3 eric7/Unittest/Interfaces/UnittestExecutor.py
--- a/eric7/Unittest/Interfaces/UnittestExecutor.py Thu May 12 09:00:35 2022 +0200
+++ b/eric7/Unittest/Interfaces/UnittestExecutor.py Fri May 13 17:23:21 2022 +0200
@@ -10,10 +10,13 @@
import contextlib
import json
import os
+import re
-from PyQt6.QtCore import QProcess
+from PyQt6.QtCore import pyqtSlot, QProcess
-from .UTExecutorBase import UTExecutorBase
+from EricNetwork.EricJsonStreamReader import EricJsonReader
+
+from .UTExecutorBase import UTExecutorBase, UTTestResult, ResultCategory
class UnittestExecutor(UTExecutorBase):
@@ -25,6 +28,33 @@
runner = os.path.join(os.path.dirname(__file__), "UnittestRunner.py")
+ def __init__(self, testWidget):
+ """
+ Constructor
+
+ @param testWidget reference to the unit test widget
+ @type UnittestWidget
+ """
+ super().__init__(testWidget)
+
+ self.__statusCategoryMapping = {
+ "failure": ResultCategory.FAIL,
+ "error": ResultCategory.FAIL,
+ "skipped": ResultCategory.SKIP,
+ "expected failure": ResultCategory.OK,
+ "unexpected success": ResultCategory.FAIL,
+ "success": ResultCategory.OK,
+ }
+
+ self.__statusDisplayMapping = {
+ "failure": self.tr("Failure"),
+ "error": self.tr("Error"),
+ "skipped": self.tr("Skipped"),
+ "expected failure": self.tr("Expected Failure"),
+ "unexpected success": self.tr("Unexpected Success"),
+ "success": self.tr("Success"),
+ }
+
def getVersions(self, interpreter):
"""
Public method to get the test framework version and version information
@@ -55,9 +85,125 @@
@type UTTestConfig
@return list of process arguments
@rtype list of str
- @exception NotImplementedError this method needs to be implemented by
- derived classes
+ """
+ args = [
+ UnittestExecutor.runner,
+ "runtest",
+ self.reader.address(),
+ str(self.reader.port()),
+ ]
+
+ if config.discover:
+ args.extend([
+ "discover",
+ "--start-directory",
+ config.discoveryStart,
+ ])
+
+ if config.failFast:
+ args.append("--failfast")
+
+ if config.collectCoverage:
+ args.append("--cover")
+ if config.eraseCoverage:
+ args.append("--cover-erase")
+
+ if config.testFilename and config.testName:
+ args.append(config.testFilename)
+ args.append(config.testName)
+
+ return args
+
+ def start(self, config, pythonpath):
+ """
+ Public method to start the testing process.
+
+ @param config configuration for the test execution
+ @type UTTestConfig
+ @param pythonpath list of directories to be added to the Python path
+ @type list of str
+ """
+ self.reader = EricJsonReader(name="Unittest Reader", parent=self)
+ self.reader.dataReceived.connect(self.__processData)
+
+ super().start(config, pythonpath)
+
+ def finished(self):
+ """
+ Public method handling the unit test process been finished.
+
+ This method should read the results (if necessary) and emit the signal
+ testFinished.
+ """
+ self.reader.close()
+
+ output = self.readAllOutput()
+ self.testFinished.emit([], output)
+
+ @pyqtSlot(object)
+ def __processData(self, data):
"""
- raise NotImplementedError
+ Private slot to process the received data.
+
+ @param data data object received
+ @type dict
+ """
+ # error collecting tests
+ if data["event"] == "collecterror":
+ self.collectError.emit([("", data["error"])])
+
+ # tests collected
+ elif data["event"] == "collected":
+ self.collected.emit([
+ (t["id"], t["name"], t["description"]) for t in data["tests"]
+ ])
+
+ # test started
+ elif data["event"] == "started":
+ self.startTest.emit(
+ (data["id"], data["name"], data["description"])
+ )
- return []
+ # test result
+ elif data["event"] == "result":
+ fn, ln = None, None
+ tracebackLines = []
+ if "traceback" in data:
+ # get the error info
+ tracebackLines = data["traceback"].splitlines()
+ # find the last entry matching the pattern
+ for index in range(len(tracebackLines) - 1, -1, -1):
+ fmatch = re.search(r'File "(.*?)", line (\d*?),.*',
+ tracebackLines[index])
+ if fmatch:
+ break
+ if fmatch:
+ fn, ln = fmatch.group(1, 2)
+
+ if "shortmsg" in data:
+ message = data["shortmsg"]
+ elif tracebackLines:
+ message = tracebackLines[-1].split(":", 1)[1].strip()
+ else:
+ message = ""
+
+ self.testResult.emit(UTTestResult(
+ category=self.__statusCategoryMapping[data["status"]],
+ status=self.__statusDisplayMapping[data["status"]],
+ name=data["name"],
+ id=data["id"],
+ description=data["description"],
+ message=message,
+ extra=tracebackLines,
+ duration=data["duration_ms"],
+ filename=fn,
+ lineno=ln,
+ ))
+
+ # test run finished
+ elif data["event"] == "finished":
+ self.testRunFinished.emit(data["tests"], data["duration_s"])
+
+ # coverage data
+ elif data["event"] == "coverage":
+ self.coverageDataSaved.emit(data["file"])
