Dont use SIGTSTP and SIGCONT in the test as they are very flaky
This commit is contained in:
parent
d5df301317
commit
7e3bd8586f
@ -1,8 +1,8 @@
|
|||||||
import termios
|
import termios
|
||||||
from ctypes import Array, c_ubyte
|
from ctypes import Array, c_ubyte
|
||||||
from typing import (
|
from typing import (
|
||||||
Any, AnyStr, Callable, Dict, List, NamedTuple, NewType, Optional, Tuple,
|
Any, AnyStr, Callable, Dict, List, NewType, Optional, Tuple, TypedDict,
|
||||||
TypedDict, Union
|
Union
|
||||||
)
|
)
|
||||||
|
|
||||||
from kitty.boss import Boss
|
from kitty.boss import Boss
|
||||||
@ -10,6 +10,7 @@ from kitty.fonts import FontFeature
|
|||||||
from kitty.fonts.render import FontObject
|
from kitty.fonts.render import FontObject
|
||||||
from kitty.marks import MarkerFunc
|
from kitty.marks import MarkerFunc
|
||||||
from kitty.options.types import Options
|
from kitty.options.types import Options
|
||||||
|
from kitty.types import SignalInfo
|
||||||
|
|
||||||
# Constants {{{
|
# Constants {{{
|
||||||
CLD_KILLED: int
|
CLD_KILLED: int
|
||||||
@ -1398,17 +1399,6 @@ def random_unix_socket() -> int:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class SignalInfo(NamedTuple):
|
|
||||||
si_signo: int
|
|
||||||
si_code: int
|
|
||||||
si_pid: int
|
|
||||||
si_uid: int
|
|
||||||
si_addr: int
|
|
||||||
si_status: int
|
|
||||||
sival_int: int
|
|
||||||
sival_ptr: int
|
|
||||||
|
|
||||||
|
|
||||||
def read_signals(fd: int, callback: Callable[[SignalInfo], None]) -> None:
|
def read_signals(fd: int, callback: Callable[[SignalInfo], None]) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@ -38,6 +38,17 @@ class WindowGeometry(NamedTuple):
|
|||||||
spaces: Edges = Edges()
|
spaces: Edges = Edges()
|
||||||
|
|
||||||
|
|
||||||
|
class SignalInfo(NamedTuple):
|
||||||
|
si_signo: int
|
||||||
|
si_code: int
|
||||||
|
si_pid: int
|
||||||
|
si_uid: int
|
||||||
|
si_addr: int
|
||||||
|
si_status: int
|
||||||
|
sival_int: int
|
||||||
|
sival_ptr: int
|
||||||
|
|
||||||
|
|
||||||
def mod_to_names(mods: int) -> Iterator[str]:
|
def mod_to_names(mods: int) -> Iterator[str]:
|
||||||
from .fast_data_types import (
|
from .fast_data_types import (
|
||||||
GLFW_MOD_ALT, GLFW_MOD_CAPS_LOCK, GLFW_MOD_CONTROL, GLFW_MOD_HYPER,
|
GLFW_MOD_ALT, GLFW_MOD_CAPS_LOCK, GLFW_MOD_CONTROL, GLFW_MOD_HYPER,
|
||||||
|
|||||||
@ -10,10 +10,10 @@ import subprocess
|
|||||||
import tempfile
|
import tempfile
|
||||||
import time
|
import time
|
||||||
|
|
||||||
from kitty.constants import is_macos, kitty_exe
|
from kitty.constants import kitty_exe
|
||||||
from kitty.fast_data_types import (
|
from kitty.fast_data_types import (
|
||||||
CLD_CONTINUED, CLD_EXITED, CLD_KILLED, CLD_STOPPED, get_options,
|
CLD_EXITED, CLD_KILLED, get_options, has_sigqueue, install_signal_handlers,
|
||||||
install_signal_handlers, read_signals, remove_signal_handlers
|
read_signals, remove_signal_handlers, sigqueue
|
||||||
)
|
)
|
||||||
|
|
||||||
from . import BaseTest
|
from . import BaseTest
|
||||||
@ -65,44 +65,58 @@ import os, json; from kitty.utils import *; from kitty.fast_data_types import ge
|
|||||||
|
|
||||||
def test_signal_handling(self):
|
def test_signal_handling(self):
|
||||||
expecting_code = 0
|
expecting_code = 0
|
||||||
|
expecting_signal = 0
|
||||||
|
expecting_value = 0
|
||||||
found_signal = False
|
found_signal = False
|
||||||
|
|
||||||
def handle_signal(siginfo):
|
def handle_signal(siginfo):
|
||||||
nonlocal found_signal
|
nonlocal found_signal
|
||||||
self.ae(siginfo.si_signo, signal.SIGCHLD)
|
if expecting_signal:
|
||||||
self.ae(siginfo.si_code, expecting_code)
|
self.ae(siginfo.si_signo, expecting_signal)
|
||||||
|
if expecting_code is not None:
|
||||||
|
self.ae(siginfo.si_code, expecting_code)
|
||||||
|
self.ae(siginfo.sival_int, expecting_value)
|
||||||
if expecting_code in (CLD_EXITED, CLD_KILLED):
|
if expecting_code in (CLD_EXITED, CLD_KILLED):
|
||||||
p.wait(1)
|
p.wait(1)
|
||||||
p.stdin.close()
|
p.stdin.close()
|
||||||
found_signal = True
|
found_signal = True
|
||||||
|
|
||||||
def t(signal, q):
|
def assert_signal():
|
||||||
nonlocal expecting_code, found_signal
|
nonlocal found_signal
|
||||||
expecting_code = q
|
|
||||||
found_signal = False
|
found_signal = False
|
||||||
|
st = time.monotonic()
|
||||||
|
while time.monotonic() - st < 5:
|
||||||
|
for (fd, event) in poll.poll(10):
|
||||||
|
if fd == signal_read_fd:
|
||||||
|
read_signals(signal_read_fd, handle_signal)
|
||||||
|
if found_signal:
|
||||||
|
break
|
||||||
|
self.assertTrue(found_signal, f'Failed to to get SIGCHLD for signal {signal}')
|
||||||
|
|
||||||
|
def t(signal, q, expecting_sig=signal.SIGCHLD):
|
||||||
|
nonlocal expecting_code, found_signal, expecting_signal
|
||||||
|
expecting_code = q
|
||||||
|
expecting_signal = expecting_sig.value
|
||||||
if signal is not None:
|
if signal is not None:
|
||||||
p.send_signal(signal)
|
p.send_signal(signal)
|
||||||
if q is not None:
|
assert_signal()
|
||||||
st = time.monotonic()
|
|
||||||
while time.monotonic() - st < 5:
|
|
||||||
for (fd, event) in poll.poll(10):
|
|
||||||
if fd == signal_read_fd:
|
|
||||||
read_signals(signal_read_fd, handle_signal)
|
|
||||||
if found_signal:
|
|
||||||
break
|
|
||||||
self.assertTrue(found_signal, f'Failed to to get SIGCHLD for signal {signal}')
|
|
||||||
|
|
||||||
poll = select.poll()
|
poll = select.poll()
|
||||||
p = subprocess.Popen([kitty_exe(), '+runpy', 'while True:\n x=2+2'], stderr=subprocess.DEVNULL, stdin=subprocess.PIPE)
|
p = subprocess.Popen([kitty_exe(), '+runpy', 'input()'], stderr=subprocess.DEVNULL, stdin=subprocess.PIPE)
|
||||||
signal_read_fd = install_signal_handlers(signal.SIGCHLD)[0]
|
signal_read_fd = install_signal_handlers(signal.SIGCHLD, signal.SIGUSR1)[0]
|
||||||
try:
|
try:
|
||||||
poll.register(signal_read_fd, select.POLLIN)
|
poll.register(signal_read_fd, select.POLLIN)
|
||||||
t(signal.SIGTSTP, CLD_STOPPED)
|
|
||||||
# macOS doesnt send SIGCHLD for SIGCONT. This is not required by POSIX sadly
|
|
||||||
t(signal.SIGCONT, None if is_macos else CLD_CONTINUED)
|
|
||||||
t(signal.SIGINT, CLD_KILLED)
|
t(signal.SIGINT, CLD_KILLED)
|
||||||
p = subprocess.Popen([kitty_exe(), '+runpy', 'input()'], stderr=subprocess.DEVNULL, stdin=subprocess.PIPE)
|
p = subprocess.Popen([kitty_exe(), '+runpy', 'input()'], stderr=subprocess.DEVNULL, stdin=subprocess.PIPE)
|
||||||
p.stdin.close()
|
p.stdin.close()
|
||||||
t(None, os.CLD_EXITED)
|
t(None, os.CLD_EXITED)
|
||||||
|
expecting_code = None
|
||||||
|
expecting_signal = signal.SIGUSR1.value
|
||||||
|
os.kill(os.getpid(), signal.SIGUSR1)
|
||||||
|
assert_signal()
|
||||||
|
expecting_value = 17 if has_sigqueue else 0
|
||||||
|
sigqueue(os.getpid(), signal.SIGUSR1.value, expecting_value)
|
||||||
|
assert_signal()
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
remove_signal_handlers()
|
remove_signal_handlers()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user