vibefpmgui.window.testing

  1"""Test and Run: listing the tests or programs, running them, in parallel too.
  2
  3Each one runs through the marker wrapper (`vibefpmgui.test_runner`), so the
  4Tests and Run panels know where each one starts and ends.
  5"""
  6
  7from __future__ import annotations
  8
  9import fnmatch
 10import json
 11import os
 12import time
 13from pathlib import Path
 14
 15from qtpy.QtCore import QProcessEnvironment
 16from qtpy.QtWidgets import QMessageBox
 17
 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
 29
 30
 31class TestingMixin:
 32    """Test and Run (a part of `vibefpmgui.main_window.MainWindow`)."""
 33
 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
 57
 58    def _marker_monitor(self):
 59        return self.run_monitor if getattr(self, "_marker_kind", "test") == "run" else self.test_monitor
 60
 61    def run_tests(self):
 62        self._start_marked("test")
 63
 64    def run_programs(self):
 65        self._start_marked("run")
 66
 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():
 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:
 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)
 99
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:
105            monitor.finish(code, stopped=stopped)
106            self.show_panel(self.build_monitor)
107            return
108        if names == []:
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:
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
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:
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:
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:
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"]:
221            message = f"{title}: {counts['Passed']} {monitor.word('passed')}, {counts['Not run']} not run"
222        elif total == 1:
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:
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 ""