|
1 # -*- coding: utf-8 -*- |
|
2 |
|
3 # Copyright (c) 2022 Detlev Offenbach <detlev@die-offenbachs.de> |
|
4 # |
|
5 |
|
6 """ |
|
7 Module implementing the executor base class for the various testing frameworks |
|
8 and supporting classes. |
|
9 """ |
|
10 |
|
11 import os |
|
12 from dataclasses import dataclass |
|
13 from enum import IntEnum |
|
14 |
|
15 from PyQt6.QtCore import pyqtSignal, QObject, QProcess, QProcessEnvironment |
|
16 |
|
17 import Preferences |
|
18 |
|
19 |
|
20 class TestResultCategory(IntEnum): |
|
21 """ |
|
22 Class defining the supported result categories. |
|
23 """ |
|
24 |
|
25 RUNNING = 0 |
|
26 FAIL = 1 |
|
27 OK = 2 |
|
28 SKIP = 3 |
|
29 PENDING = 4 |
|
30 |
|
31 |
|
32 @dataclass |
|
33 class TestResult: |
|
34 """ |
|
35 Class containing the test result data. |
|
36 """ |
|
37 |
|
38 category: TestResultCategory # result category |
|
39 status: str # test status |
|
40 name: str # test name |
|
41 id: str # test id |
|
42 description: str = "" # short description of test |
|
43 message: str = "" # short result message |
|
44 extra: list = None # additional information text |
|
45 duration: float = None # test duration |
|
46 filename: str = None # file name of a failed test |
|
47 lineno: int = None # line number of a failed test |
|
48 subtestResult: bool = False # flag indicating the result of a subtest |
|
49 |
|
50 |
|
51 @dataclass |
|
52 class TestConfig: |
|
53 """ |
|
54 Class containing the test run configuration. |
|
55 """ |
|
56 |
|
57 interpreter: str # path of the Python interpreter |
|
58 discover: bool # auto discovery flag |
|
59 discoveryStart: str # start directory for auto discovery |
|
60 testFilename: str # name of the test script |
|
61 testName: str # name of the test function |
|
62 failFast: bool # stop on first fail |
|
63 failedOnly: bool # run failed tests only |
|
64 collectCoverage: bool # coverage collection flag |
|
65 eraseCoverage: bool # erase coverage data first |
|
66 coverageFile: str # name of the coverage data file |
|
67 |
|
68 |
|
69 class TestExecutorBase(QObject): |
|
70 """ |
|
71 Base class for test framework specific implementations. |
|
72 |
|
73 @signal collected(list of tuple of (str, str, str)) emitted after all tests |
|
74 have been collected. Tuple elements are the test id, the test name and |
|
75 a short description of the test. |
|
76 @signal collectError(list of tuple of (str, str)) emitted when errors |
|
77 are encountered during test collection. Tuple elements are the |
|
78 test name and the error message. |
|
79 @signal startTest(tuple of (str, str, str) emitted before tests are run. |
|
80 Tuple elements are test id, test name and short description. |
|
81 @signal testResult(TestResult) emitted when a test result is ready |
|
82 @signal testFinished(list, str) emitted when the test has finished. |
|
83 The elements are the list of test results and the captured output |
|
84 of the test worker (if any). |
|
85 @signal testRunAboutToBeStarted() emitted just before the test run will |
|
86 be started. |
|
87 @signal testRunFinished(int, float) emitted when the test run has finished. |
|
88 The elements are the number of tests run and the duration in seconds |
|
89 @signal stop() emitted when the test process is being stopped. |
|
90 @signal coverageDataSaved(str) emitted after the coverage data was saved. |
|
91 The element is the absolute path of the coverage data file. |
|
92 """ |
|
93 |
|
94 collected = pyqtSignal(list) |
|
95 collectError = pyqtSignal(list) |
|
96 startTest = pyqtSignal(tuple) |
|
97 testResult = pyqtSignal(TestResult) |
|
98 testFinished = pyqtSignal(list, str) |
|
99 testRunAboutToBeStarted = pyqtSignal() |
|
100 testRunFinished = pyqtSignal(int, float) |
|
101 stop = pyqtSignal() |
|
102 coverageDataSaved = pyqtSignal(str) |
|
103 |
|
104 module = "" |
|
105 name = "" |
|
106 runner = "" |
|
107 |
|
108 def __init__(self, testWidget): |
|
109 """ |
|
110 Constructor |
|
111 |
|
112 @param testWidget reference to the unit test widget |
|
113 @type TestingWidget |
|
114 """ |
|
115 super().__init__(testWidget) |
|
116 |
|
117 self.__process = None |
|
118 |
|
119 @classmethod |
|
120 def isInstalled(cls, interpreter): |
|
121 """ |
|
122 Class method to check whether a test framework is installed. |
|
123 |
|
124 The test is performed by checking, if a module loader can found. |
|
125 |
|
126 @param interpreter interpreter to be used for the test |
|
127 @type str |
|
128 @return flag indicating the test framework module is installed |
|
129 @rtype bool |
|
130 """ |
|
131 if cls.runner: |
|
132 proc = QProcess() |
|
133 proc.start(interpreter, [cls.runner, "installed"]) |
|
134 if proc.waitForFinished(3000): |
|
135 exitCode = proc.exitCode() |
|
136 return exitCode == 0 |
|
137 |
|
138 return False |
|
139 |
|
140 def getVersions(self, interpreter): |
|
141 """ |
|
142 Public method to get the test framework version and version information |
|
143 of its installed plugins. |
|
144 |
|
145 @param interpreter interpreter to be used for the test |
|
146 @type str |
|
147 @return dictionary containing the framework name and version and the |
|
148 list of available plugins with name and version each |
|
149 @rtype dict |
|
150 @exception NotImplementedError this method needs to be implemented by |
|
151 derived classes |
|
152 """ |
|
153 raise NotImplementedError |
|
154 |
|
155 return {} |
|
156 |
|
157 def hasCoverage(self, interpreter): |
|
158 """ |
|
159 Public method to get the test framework version and version information |
|
160 of its installed plugins. |
|
161 |
|
162 @param interpreter interpreter to be used for the test |
|
163 @type str |
|
164 @return flag indicating the availability of coverage functionality |
|
165 @rtype bool |
|
166 @exception NotImplementedError this method needs to be implemented by |
|
167 derived classes |
|
168 """ |
|
169 raise NotImplementedError |
|
170 |
|
171 return False |
|
172 |
|
173 def createArguments(self, config): |
|
174 """ |
|
175 Public method to create the arguments needed to start the test process. |
|
176 |
|
177 @param config configuration for the test execution |
|
178 @type TestConfig |
|
179 @return list of process arguments |
|
180 @rtype list of str |
|
181 @exception NotImplementedError this method needs to be implemented by |
|
182 derived classes |
|
183 """ |
|
184 raise NotImplementedError |
|
185 |
|
186 return [] |
|
187 |
|
188 def _prepareProcess(self, workDir, pythonpath): |
|
189 """ |
|
190 Protected method to prepare a process object to be started. |
|
191 |
|
192 @param workDir working directory |
|
193 @type str |
|
194 @param pythonpath list of directories to be added to the Python path |
|
195 @type list of str |
|
196 @return prepared process object |
|
197 @rtype QProcess |
|
198 """ |
|
199 process = QProcess(self) |
|
200 process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels) |
|
201 process.setWorkingDirectory(workDir) |
|
202 process.finished.connect(self.finished) |
|
203 if pythonpath: |
|
204 env = QProcessEnvironment.systemEnvironment() |
|
205 currentPythonPath = env.value("PYTHONPATH", None) |
|
206 newPythonPath = os.pathsep.join(pythonpath) |
|
207 if currentPythonPath: |
|
208 newPythonPath += os.pathsep + currentPythonPath |
|
209 env.insert("PYTHONPATH", newPythonPath) |
|
210 process.setProcessEnvironment(env) |
|
211 |
|
212 return process |
|
213 |
|
214 def start(self, config, pythonpath): |
|
215 """ |
|
216 Public method to start the testing process. |
|
217 |
|
218 @param config configuration for the test execution |
|
219 @type TestConfig |
|
220 @param pythonpath list of directories to be added to the Python path |
|
221 @type list of str |
|
222 @exception RuntimeError raised if the the testing process did not start |
|
223 """ |
|
224 workDir = ( |
|
225 config.discoveryStart |
|
226 if config.discover |
|
227 else os.path.dirname(config.testFilename) |
|
228 ) |
|
229 self.__process = self._prepareProcess(workDir, pythonpath) |
|
230 testArgs = self.createArguments(config) |
|
231 self.testRunAboutToBeStarted.emit() |
|
232 self.__process.start(config.interpreter, testArgs) |
|
233 running = self.__process.waitForStarted() |
|
234 if not running: |
|
235 raise RuntimeError |
|
236 |
|
237 def finished(self): |
|
238 """ |
|
239 Public method handling the unit test process been finished. |
|
240 |
|
241 This method should read the results (if necessary) and emit the signal |
|
242 testFinished. |
|
243 |
|
244 @exception NotImplementedError this method needs to be implemented by |
|
245 derived classes |
|
246 """ |
|
247 raise NotImplementedError |
|
248 |
|
249 def readAllOutput(self, process=None): |
|
250 """ |
|
251 Public method to read all output of the test process. |
|
252 |
|
253 @param process reference to the process object |
|
254 @type QProcess |
|
255 @return test process output |
|
256 @rtype str |
|
257 """ |
|
258 if process is None: |
|
259 process = self.__process |
|
260 output = ( |
|
261 str( |
|
262 process.readAllStandardOutput(), |
|
263 Preferences.getSystem("IOEncoding"), |
|
264 "replace", |
|
265 ).strip() |
|
266 if process |
|
267 else "" |
|
268 ) |
|
269 return output |
|
270 |
|
271 def stopIfRunning(self): |
|
272 """ |
|
273 Public method to stop the testing process, if it is running. |
|
274 """ |
|
275 if self.__process and self.__process.state() == QProcess.ProcessState.Running: |
|
276 self.__process.terminate() |
|
277 self.__process.waitForFinished(2000) |
|
278 self.__process.kill() |
|
279 self.__process.waitForFinished(3000) |
|
280 |
|
281 self.stop.emit() |