Something went wrong. Try again.
paimon The best travel companion ever!
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136"""Shell output generators and readers shared by tool tests.
The commands here are Python one-liners rather than shell loops: the shell toolruns /bin/sh or bash on POSIX and cmd.exe on Windows, and the only syntax thetwo share is a quoted program with a quoted argument. What the tests exerciseis what paimon does with a command's output and its process tree, which doesnot depend on who wrote the bytes."""
import osimport reimport signalimport subprocessimport sysimport unittestfrom pathlib import Path
# For the few tests that are about the POSIX shell itself ($0, bashisms).posix_shell = unittest.skipUnless(os.name == "posix", "needs a POSIX shell")
# Stops SIGTERM from ending the process, so a test reaches the forced kill.# A no-op on Windows, where nothing delivers SIGTERM to begin with.IGNORE_TERM = "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN); "
_SLEEPER = "import time; time.sleep(30)"
def python_command(code: str) -> str: """``code`` as a command line both sh and cmd.exe run the same way.
The code goes inside double quotes, so it must hold none itself, and no ``$``, backtick or ``%`` either; a backslash escape such as ``\\n`` passes through both shells untouched and is Python's to interpret. """ assert not set(code) & set('"$`%'), code return f'"{sys.executable}" -c "{code}"'
def filler(lines: int, payload: str, newline: bool = True, *, last: str = "", exit_code: int = 0) -> str: """A command printing ``payload`` ``lines`` times, then ``last`` on a line of its own, and exiting with ``exit_code``.
Written as bytes: text mode would turn every newline into CRLF on Windows and encode with the console's code page rather than UTF-8. """ chunk = ascii(payload + ("\n" if newline else "")) code = f"import sys; w = sys.stdout.buffer.write; w({chunk}.encode('utf-8') * {lines}); " if last: code += f"w({ascii(last + chr(10))}.encode('utf-8')); " return python_command(code + f"sys.exit({exit_code})")
def printer(text: str, *, then_sleep: bool = False, exit_code: int = 0, ignore_term: bool = False) -> str: """A command that prints ``text`` on a line, then sleeps or exits.""" code = IGNORE_TERM if ignore_term else "" code += (f"import sys; sys.stdout.buffer.write({ascii(text + chr(10))}.encode('utf-8')); " "sys.stdout.flush(); ") code += _SLEEPER if then_sleep else f"sys.exit({exit_code})" return python_command(code)
def sleeper(*, ignore_term: bool = False) -> str: """A command that just outstays the test.""" return python_command((IGNORE_TERM if ignore_term else "") + _SLEEPER)
def spawner(*, then_sleep: bool, ignore_term: bool = False, own_pid: bool = False) -> str: """A command that starts a long-lived child and prints ``pid <n>``.
The pid is the child's — the descendant a tree kill has to reach — or the command's own with ``own_pid``. With ``then_sleep`` the command stays alive beside its child; without, it exits and leaves the child behind. """ code = IGNORE_TERM if ignore_term else "" code += ("import os, subprocess, sys, time; " f"child = subprocess.Popen([sys.executable, '-c', '{_SLEEPER}']); " f"print('pid', {'os.getpid()' if own_pid else 'child.pid'}, flush=True); ") if then_sleep: code += "time.sleep(30)" return python_command(code)
def pid_from(output: str) -> int: match = re.search(r"pid (\d+)", output) assert match, f"no pid in output: {output!r}" return int(match.group(1))
def alive(pid: int) -> bool: """Whether a process with this pid exists.
Never ``os.kill(pid, 0)`` on Windows: there any signal number that is not a console event is passed to TerminateProcess as the exit code, so the probe would kill the process it was asking about. """ if os.name == "nt": import ctypes from ctypes import wintypes
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.OpenProcess.restype = wintypes.HANDLE handle = kernel32.OpenProcess(0x1000, False, pid) # QUERY_LIMITED_INFORMATION if not handle: return False try: code = wintypes.DWORD() kernel32.GetExitCodeProcess(wintypes.HANDLE(handle), ctypes.byref(code)) return code.value == 259 # STILL_ACTIVE finally: kernel32.CloseHandle(wintypes.HANDLE(handle)) try: os.kill(pid, 0) except (ProcessLookupError, PermissionError): return False return True
def reap(pid: int) -> None: """Kill a process a test started, whatever its assertions did.""" if os.name == "nt": subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return try: os.kill(pid, signal.SIGKILL) except (ProcessLookupError, PermissionError): pass
def overflow_path(result: str) -> Path: match = re.search(r"full output: (\S+?)\]", result) assert match, f"no overflow path in result: {result[-300:]!r}" return Path(match.group(1))