Coverage for vibefpmgui/parallel

1"""Running tests in parallel (Build options → Parallel tests).

3fpm runs tests one after another, so for a parallel run fpm only builds

4them: ``fpm test --runner "test_runner.py --collect"`` prints each test's

5command line instead of running it (see `vibefpmgui.test_runner`). Then

6`ParallelTests` runs those commands itself, a few at a time, each in its own

7process with its own output, so the outputs never mix.

8"""

10import codecs

11import time

12from collections.abc import Callable

14from qtpy.QtCore import QObject, QProcess, QProcessEnvironment, Signal

17class ParallelTests(QObject):

18 """Run ``(name, argv)`` commands, at most ``workers`` at a time. Signals

19 say when each starts, what it prints and how it ended; `done` comes once

20 all have ended, or once the ones running when `stop` was called have."""

22 started = Signal(str) # name

23 output = Signal(str, str) # name, text

24 finished = Signal(str, int, float, bool) # name, exit code, seconds, stopped

25 done = Signal(bool) # stopped

27 def __init__(self, commands: list[tuple[str, list[str]]], workers: int, cwd: str,

28 env: QProcessEnvironment, inner: list[str], kill: Callable[[QProcess], None],

29 parent=None):

30 """``inner``: a runner put in front of each command (fpm's --runner,

31 e.g. ``mpirun -np 2``); ``kill`` stops a process and what it started."""

32 super().__init__(parent)

33 self.queue = list(commands)

34 self.workers = max(1, workers)

35 self.cwd, self.env, self.inner, self.kill = cwd, env, inner, kill

36 self.running: dict[str, QProcess] = {}

37 self.stopping = False

38 self._started_at: dict[str, float] = {}

39 self._decoders: dict[str, codecs.IncrementalDecoder] = {}

41 def start(self):

42 self._fill()

44 def stop(self):

45 """Run no more, and kill the ones running (`done` follows)."""

46 self.stopping = True

47 self.queue.clear()

48 for process in list(self.running.values()):

49 self.kill(process)

50 if not self.running: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 self.done.emit(True)

53 def wait(self, msecs: int = 3000):

54 for process in list(self.running.values()):

55 process.waitForFinished(msecs)

57 def command(self, argv: list[str]) -> list[str]:

58 return self.inner + argv

60 def _fill(self):

61 while self.queue and len(self.running) < self.workers and not self.stopping:

62 name, argv = self.queue.pop(0)

63 self._launch(name, self.command(argv))

64 if not self.running and not self.queue:

65 self.done.emit(self.stopping)

67 def _launch(self, name: str, command: list[str]):

68 process = QProcess(self)

69 process.setWorkingDirectory(self.cwd)

70 process.setProcessEnvironment(self.env)

71 process.setProcessChannelMode(QProcess.MergedChannels)

72 # (Lambdas are fine here: a QProcess's signals come in the GUI thread.)

73 process.readyReadStandardOutput.connect(lambda: self._read(name, process))

74 process.finished.connect(lambda code, status: self._ended(name, process, code, status))

75 process.errorOccurred.connect(lambda error: self._error(name, process, error, command[0]))

76 self.running[name] = process

77 self._started_at[name] = time.monotonic()

78 self._decoders[name] = codecs.getincrementaldecoder("utf-8")(errors="replace")

79 self.started.emit(name)

80 process.start(command[0], command[1:])

82 def _read(self, name: str, process: QProcess, final: bool = False):

83 text = self._decoders[name].decode(bytes(process.readAllStandardOutput()), final=final)

84 if text:

85 text = text.replace("\r\n", "\n")

86 # (Progress lines redrawn with \r: only their last state.)

87 text = "\n".join(line.rsplit("\r", 1)[-1] for line in text.split("\n"))

88 self.output.emit(name, text)

90 def _ended(self, name: str, process: QProcess, code: int, status):

91 if name not in self.running: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 return

93 self._read(name, process, final=True)

94 stopped = self.stopping

95 if status == QProcess.CrashExit and not stopped: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 code = code or -1 # (a crash: the exit code isn't the program's)

97 self._finish(name, process, code, stopped)

99 def _error(self, name: str, process: QProcess, error, program: str):

100 if error == QProcess.FailedToStart and name in self.running:

101 self.output.emit(name, f"could not run {program}: {process.errorString()}\n")

102 self._finish(name, process, 127, False)

103

104 def _finish(self, name: str, process: QProcess, code: int, stopped: bool):

105 del self.running[name]

106 seconds = time.monotonic() - self._started_at.pop(name, time.monotonic())

107 self._decoders.pop(name, None)

108 process.deleteLater()

109 self.finished.emit(name, code, seconds, stopped)

110 self._fill()

Coverage for vibefpmgui/parallel_tests.py: 89%

76 statements