Coverage for vibefpmgui/window/testing.py: 86%

1"""Test and Run: listing the tests or programs, running them, in parallel too.

3Each one runs through the marker wrapper (`vibefpmgui.test_runner`), so the

4Tests and Run panels know where each one starts and ends.

5"""

7from __future__ import annotations

9import fnmatch

10import json

11import os

12import time

13from pathlib import Path

15from qtpy.QtCore import QProcessEnvironment

16from qtpy.QtWidgets import QMessageBox

18from .. import ansi

19from .. import coverage as cov

20from ..build_monitor import BuildMonitor

21from ..compat import join_args, split_args

22from ..coverage_panel import CoveragePanel

23from ..fpm_args import runner_command, split_runner

24from ..parallel_tests import ParallelTests

25from ..processes import kill_process_tree

26from ..run_monitor import RunMonitor

27from ..test_monitor import EXE_RE, MARKER_RE, TestMonitor, parse_test_list

28from ..test_runner import split_runner as split_inner_runner

31class TestingMixin:

32 """Test and Run (a part of `vibefpmgui.main_window.MainWindow`)."""

34 def _build_run_panels(self):

35 """The Build, Tests, Run and Coverage panels."""

36 self.build_monitor = BuildMonitor()

37 self.test_monitor = TestMonitor()

38 self.test_monitor.test_selected.connect(self.show_test_output)

39 self.test_monitor.debug_requested.connect(lambda name: self.debug_named("test", name))

40 self.test_monitor.build_percent = lambda: self.build_monitor.percent

41 self.build_monitor.details_requested.connect(self.show_build_details)

42 self.build_monitor.open_file_requested.connect(self.open_file)

43 self.build_monitor.compile_command_requested.connect(self.show_compile_command)

44 self.run_monitor = RunMonitor()

45 self.run_monitor.debug_requested.connect(lambda name: self.debug_named("program", name))

46 self.run_monitor.test_selected.connect(

47 lambda name: self._show_marked_output(self.run_monitor, name))

48 self.run_monitor.build_percent = lambda: self.build_monitor.percent

49 self.coverage_panel = CoveragePanel()

50 self.coverage_panel.open_requested.connect(

51 lambda path, line: self.navigate_to(path, line, 0, preview=True))

52 self.coverage_panel.tests_check.setChecked(self.settings.value("coverage_tests", False, type=bool))

53 self.coverage_panel.tests_check.toggled.connect(lambda on: self.settings.setValue("coverage_tests", on))

54 self.coverage: dict[Path, cov.FileCoverage] = {} # the last test run's (see _collect_coverage)

55 self._coverage_started = 0.0 # when that run started

56 self._coverage_collector: cov.Collector | None = None

58 def _marker_monitor(self):

59 return self.run_monitor if getattr(self, "_marker_kind", "test") == "run" else self.test_monitor

61 def run_tests(self):

62 self._start_marked("test")

64 def run_programs(self):

65 self._start_marked("run")

67 def _start_marked(self, kind: str):

68 """Test or Run in two steps: "fpm <kind> --list" (the names), then

69 "fpm <kind> --runner <our wrapper>", which builds first (shown in the

70 Build tab) and then runs each one; the wrapper marks where each starts

71 and ends, so the Tests / Run tab can follow them."""

72 if not self.manifest_path or self.busy(): 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 return

74 try:

75 list_args = self.build_arguments([kind, "--list"], True, program_args=False)

76 run_args = self.build_arguments([kind], True)

77 _, inner_runner = split_runner(split_args(self.extra_edit.text()))

78 except ValueError as exc:

79 QMessageBox.warning(self, "Invalid arguments", str(exc))

80 return

81 runner = runner_command()

82 # Our runner goes before the target names and any "-- program args".

83 cut = run_args.index("--") if "--" in run_args else len(run_args)

84 collect_args = run_args[:cut] + ["--runner", runner_command(collect=True)] + run_args[cut:]

85 run_args = run_args[:cut] + ["--runner", runner] + run_args[cut:]

86 patterns = split_args(self.target_edit.text())

87 self._marker_kind = kind

88 self._test_args = (run_args, inner_runner, patterns, collect_args)

89 description = self._build_description()

90 if inner_runner: 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true

91 description += f" · runner: {inner_runner}"

92 self.build_monitor.reset() # the build progress shown next is this run's

93 if kind == "test" and self.coverage_check.isChecked():

94 cov.clear_counts(self.manifest_path.parent / "build") # (this run's counts only)

95 self._coverage_started = time.time() # (files changed after this aren't what was compiled)

96 self._marker_monitor().begin_listing(description)

97 self.show_all_output()

98 self.run_fpm([kind], True, False, phase="list", args=list_args)

100 def _after_listing(self, code: int, stopped: bool):

101 monitor = self._marker_monitor()

102 output = "".join(self._captured)

103 names = parse_test_list(output)

104 if stopped or code != 0: 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true

105 monitor.finish(code, stopped=stopped)

106 self.show_panel(self.build_monitor)

107 return

108 if names == []: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 monitor.finish(0, no_tests=True)

110 self.show_panel(monitor)

111 self.statusBar().showMessage(monitor.word("none"))

112 return

113 run_args, inner_runner, patterns, collect_args = self._test_args

114 if names and patterns:

115 # "--list" lists every name regardless of the names given, so apply

116 # the Target names/globs ourselves, as fpm will.

117 names = [n for n in names if any(fnmatch.fnmatchcase(n, p) for p in patterns)]

118 # fpm builds first (the Build tab shows it), then runs them.

119 monitor.start(names or [], self._load_durations(self._marker_kind), building=True)

120 jobs = self.test_jobs_spin.value() if self._marker_kind == "test" else 1

121 monitor.jobs = jobs

122 if jobs > 1:

123 monitor.description += f" · {jobs} at a time"

124 monitor.update_header() # fpm only builds them and says how to run each; we run them (see _after_collect)

125 self._collected: list[tuple[str, list[str]]] = []

126 self.run_fpm(["test"], True, False, phase="collect", args=collect_args)

127 return

128 env = {"VIBEFPM_INNER_RUNNER": inner_runner} if inner_runner else {}

129 self.run_fpm([self._marker_kind], True, True, phase="marked", args=run_args, env_extra=env)

130

131 def _after_collect(self, code: int, stopped: bool):

132 """fpm has built the tests and said how to run each: run them, a few

133 at a time (Build options → Parallel tests)."""

134 if stopped or code != 0 or not self._collected: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 self.reload() # (first, as after "fpm test": then the message is the tests')

136 self._after_marked(code, stopped) # (stopped, the build failed, or no tests)

137 return

138 _run_args, inner_runner, _patterns, _collect = self._test_args

139 try:

140 inner = split_inner_runner(inner_runner)

141 except ValueError:

142 inner = []

143 expected = self._load_durations("test")

144 # The longest first (the ones never timed first of all): the run ends sooner.

145 commands = sorted(self._collected, key=lambda c: -expected.get(c[0], float("inf")))

146 env = QProcessEnvironment.systemEnvironment()

147 env.insert("NO_COLOR", "1")

148 workers = min(self.test_jobs_spin.value(), len(commands))

149 tests = ParallelTests(commands, workers, str(self.manifest_path.parent), env, inner, kill_process_tree, self)

150 tests.started.connect(self._parallel_started)

151 tests.output.connect(self._parallel_output)

152 tests.finished.connect(self._parallel_finished)

153 tests.done.connect(self._parallel_done)

154 self.parallel_tests = tests

155 self._parallel_raw: dict[str, list[str]] = {}

156 self._parallel_failed = False

157 self._append(f"\n[Running {len(commands)} tests, {workers} at a time. Each test's output is shown "

158 "here when it finishes; click a test in the Tests panel to watch it as it runs.]\n", bold=True)

159 self.stop_action.setEnabled(True)

160 self._set_project_enabled(True)

161 self.statusBar().showMessage(f"Running tests, {workers} at a time…")

162 tests.start()

163

164 def _parallel_started(self, name: str):

165 self._parallel_raw[name] = []

166 self.test_monitor.test_started(name)

167

168 def _parallel_output(self, name: str, text: str):

169 self._parallel_raw.setdefault(name, []).append(text)

170 plain = ansi.strip(text)

171 self.test_monitor.test_output(name, plain)

172 if self.test_monitor.selected == name: # shown on its own: follow it 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true

173 self.single_writer.write(plain)

174

175 def _parallel_finished(self, name: str, code: int, seconds: float, stopped: bool):

176 self.test_monitor.test_finished(name, code, seconds, stopped)

177 self._parallel_failed |= code != 0 and not stopped

178 # Its output, all in one piece: the tests' outputs never mix here.

179 status = "stopped" if stopped else "passed" if code == 0 else f"failed (exit code {code})"

180 command = self.parallel_tests.command(dict(self._collected).get(name, [name])) \

181 if self.parallel_tests is not None else [name]

182 rel = [os.path.relpath(a, self.manifest_path.parent) if os.path.isabs(a) else a for a in command]

183 self._append(f"\n── {name} · {status} · {seconds:.2f} s ── $ {join_args(rel)}\n", bold=True)

184 text = "".join(self._parallel_raw.pop(name, []))

185 if text:

186 self._append(text if text.endswith("\n") else text + "\n", colored=True)

187

188 def _parallel_done(self, stopped: bool):

189 if self.parallel_tests is None: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true

190 return

191 self.parallel_tests.deleteLater()

192 self.parallel_tests = None

193 self.stop_action.setEnabled(False)

194 self._set_project_enabled(self.graph is not None)

195 code = 1 if self._parallel_failed else 0

196 self._append("\n[tests stopped]\n" if stopped else f"\n[tests finished: exit code {code}]\n", bold=True)

197 self.reload() # (first, as after "fpm test": then the message is the tests')

198 self._after_marked(code, stopped)

199

200 def _after_marked(self, code: int, stopped: bool):

201 monitor = self._marker_monitor()

202 monitor.finish(code, stopped=stopped)

203 self._save_durations(self._marker_kind, monitor.durations())

204 if self._marker_kind == "test" and self.coverage_check.isChecked() and not stopped \

205 and not monitor.build_failed:

206 self._collect_coverage()

207 counts = monitor.counts()

208 total = len(monitor.tests)

209 title = monitor.word("title")

210 if stopped:

211 message = f"{title}: stopped"

212 elif monitor.build_failed: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true

213 message = f"{title}: the build failed (see the Build tab)"

214 elif counts["Failed"]:

215 failed = [t for t in monitor.tests if t.status == "Failed"]

216 if self._marker_kind == "run" and len(failed) == 1 and failed[0].exit_code is not None: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true

217 message = f"{title}: {failed[0].name} failed (exit code {failed[0].exit_code})"

218 else:

219 message = f"{title}: {counts['Failed']} of {total} failed"

220 elif counts["Not run"]: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true

221 message = f"{title}: {counts['Passed']} {monitor.word('passed')}, {counts['Not run']} not run"

222 elif total == 1: 222 ↛ 225line 222 didn't jump to line 225 because the condition on line 222 was always true

223 message = f"{title}: {monitor.tests[0].name} {monitor.word('passed')}"

224 else:

225 message = f"{title}: all {total} {monitor.word('passed')}"

226 self.statusBar().showMessage(message)

227

228 def _durations_key(self, kind: str = "test") -> str:

229 return self._project_key(f"{kind}_durations")

230

231 def _load_durations(self, kind: str = "test") -> dict[str, float]:

232 try:

233 data = json.loads(self.settings.value(self._durations_key(kind), "{}"))

234 return {str(k): float(v) for k, v in data.items()}

235 except (TypeError, ValueError, AttributeError):

236 return {}

237

238 def _save_durations(self, kind: str, durations: dict[str, float]):

239 if durations and self.manifest_path:

240 merged = {**self._load_durations(kind), **durations}

241 self.settings.setValue(self._durations_key(kind), json.dumps(merged))

242

243 def _test_line(self, raw: str) -> str:

244 """Pass a line of test output through the test monitor; returns the

245 text to show (with its color codes). The runner prints a blank line

246 before each marker, which is dropped along with the marker."""

247 line = ansi.strip(raw)

248 if self._test_building:

249 if MARKER_RE.search(line):

250 self._build_part_done() # the first test is starting

251 else:

252 self.build_monitor.feed_line(line) # fpm is still compiling

253 text = ""

254 if self._held_blank:

255 self._held_blank = False

256 if not MARKER_RE.search(line): # the blank was real output

257 self._marker_monitor().feed_line("")

258 text = "\n"

259 if line == "":

260 self._held_blank = True

261 return text

262 shown = self._marker_monitor().feed_line(line)

263 if shown == line:

264 shown = raw

265 return text + (shown + "\n" if shown is not None else "")

266

267 def _collect_line(self, raw: str) -> str:

268 """A line of "fpm test --runner '… --collect'": the build's output,

269 or a test's command line (taken out, kept for _after_collect)."""

270 line = ansi.strip(raw)

271 match = EXE_RE.search(line)

272 if match is None:

273 if self._test_building: 273 ↛ 275line 273 didn't jump to line 275 because the condition on line 273 was always true

274 self.build_monitor.feed_line(line)

275 return raw + "\n"

276 if self._test_building:

277 self._build_part_done()

278 try:

279 argv = json.loads(match.group(2))

280 except ValueError:

281 return raw + "\n"

282 self._collected.append((match.group(1), [str(a) for a in argv]))

283 return ""