Coverage for vibefpmgui/processes.py: 85%

1"""Ending processes: a QProcess and everything it started.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import signal 

7import subprocess 

8 

9from qtpy.QtCore import QProcess 

10 

11from .console import NO_WINDOW 

12 

13 

14def kill_process_tree(process: QProcess): 

15 """Kill a QProcess *and everything it started* (fpm's compilers, the test 

16 runner and the test program): killing only fpm would leave them running.""" 

17 pid = int(process.processId() or 0) 

18 if pid > 0: 

19 if os.name == "nt": 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true

20 subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, check=False, **NO_WINDOW) 

21 else: 

22 for child in reversed(_descendants(pid)): # deepest first 

23 try: 

24 os.kill(child, signal.SIGKILL) 

25 except OSError: 

26 pass # already gone 

27 process.kill() 

28 

29 

30def _descendants(pid: int) -> list[int]: 

31 """All processes below ``pid``, parents before children (POSIX).""" 

32 try: 

33 out = subprocess.run(["ps", "-A", "-o", "pid=", "-o", "ppid="], 

34 capture_output=True, text=True, timeout=5, check=False).stdout 

35 except (OSError, subprocess.SubprocessError): 

36 return [] 

37 children: dict[int, list[int]] = {} 

38 for line in out.splitlines(): 

39 parts = line.split() 

40 if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit(): 40 ↛ 38line 40 didn't jump to line 38 because the condition on line 40 was always true

41 children.setdefault(int(parts[1]), []).append(int(parts[0])) 

42 found, queue = [], [pid] 

43 while queue: 

44 for child in children.get(queue.pop(0), []): 

45 found.append(child) 

46 queue.append(child) 

47 return found