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