summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-cli/tests/console.py
blob: d3ea6c90b322fa60e651b4b6e0e4ca79a63e5723 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
"""
A person's Ctrl+C, delivered to a client run started by a test.

On Windows Ctrl+C is a console event, not a signal, so the run gets a hidden
console of its own and a helper attached to that console raises the event. The
run goes through a launcher that turns Ctrl+C handling back on first: a process
inherits "ignore Ctrl+C" from its parent, and whatever started the test suite
may have set it (an IDE, a CI runner, an agent's shell) where a person's console
never does — measured, the event then reached nothing at all.
"""

import subprocess
import sys

_LAUNCHER = ("import ctypes, signal, subprocess, sys\n"
             "ctypes.windll.kernel32.SetConsoleCtrlHandler(None, 0)\n"
             "signal.signal(signal.SIGINT, signal.SIG_IGN)\n"
             "sys.exit(subprocess.call(sys.argv[1:]))\n")

# os._exit: with the console swapped, flushing stdout at exit fails (code 120).
_SENDER = ("import ctypes, os, sys\n"
           "k = ctypes.windll.kernel32\n"
           "k.FreeConsole()\n"
           "ok = k.AttachConsole(int(sys.argv[1])) and k.SetConsoleCtrlHandler(None, 1)\n"
           "os._exit(0 if ok and k.GenerateConsoleCtrlEvent(0, 0) else 1)\n")


def interruptible(argv: list[str], env: dict | None) -> subprocess.Popen:
    """`argv` started so that `ctrl_c` reaches it, its output read as UTF-8."""
    extra: dict = {}
    if sys.platform == "win32":
        si = subprocess.STARTUPINFO()
        si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
        si.wShowWindow = 0                                  # SW_HIDE
        extra = {"creationflags": subprocess.CREATE_NEW_CONSOLE, "startupinfo": si}
        argv = [sys.executable, "-c", _LAUNCHER, *argv]
    return subprocess.Popen(argv, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                            text=True, encoding="utf-8", **extra)


def ctrl_c(proc: subprocess.Popen) -> None:
    if sys.platform != "win32":
        import signal
        proc.send_signal(signal.SIGINT)
        return
    subprocess.run([sys.executable, "-c", _SENDER, str(proc.pid)], check=True, timeout=30,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                   creationflags=subprocess.CREATE_NO_WINDOW)