From 5ee1f15090727c9ded0a659f01a9d057b7d9bb9f Mon Sep 17 00:00:00 2001 From: Kevin Van Brunt Date: Thu, 12 Sep 2024 14:43:29 -0400 Subject: [PATCH] Not handling SIGHUP and SIGTERM on Windows. --- CHANGELOG.md | 2 +- cmd2/argparse_custom.py | 12 ++++------- cmd2/cmd2.py | 48 +++++++++++++++++++---------------------- cmd2/decorators.py | 12 ++++------- cmd2/history.py | 6 ++---- tests/test_cmd2.py | 11 ++++++++++ 6 files changed, 44 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c638ee40..ef42ed89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ * Breaking Change * `cmd2` 2.5 supports Python 3.7+ (removed support for Python 3.6) * Bug Fixes - * Fixed issue where persistent history file was not saved upon SIGTERM and SIGHUP signals. + * Fixed issue where persistent history file was not saved upon SIGHUP and SIGTERM signals. * Enhancements * Removed dependency on `attrs` and replaced with [dataclasses](https://docs.python.org/3/library/dataclasses.html) * add `allow_clipboard` initialization parameter and attribute to disable ability to diff --git a/cmd2/argparse_custom.py b/cmd2/argparse_custom.py index 2371fa54..595737c5 100644 --- a/cmd2/argparse_custom.py +++ b/cmd2/argparse_custom.py @@ -352,8 +352,7 @@ class ChoicesProviderFuncBase(Protocol): Function that returns a list of choices in support of tab completion """ - def __call__(self) -> List[str]: - ... # pragma: no cover + def __call__(self) -> List[str]: ... # pragma: no cover @runtime_checkable @@ -362,8 +361,7 @@ class ChoicesProviderFuncWithTokens(Protocol): Function that returns a list of choices in support of tab completion and accepts a dictionary of prior arguments. """ - def __call__(self, *, arg_tokens: Dict[str, List[str]] = {}) -> List[str]: - ... # pragma: no cover + def __call__(self, *, arg_tokens: Dict[str, List[str]] = {}) -> List[str]: ... # pragma: no cover ChoicesProviderFunc = Union[ChoicesProviderFuncBase, ChoicesProviderFuncWithTokens] @@ -381,8 +379,7 @@ def __call__( line: str, begidx: int, endidx: int, - ) -> List[str]: - ... # pragma: no cover + ) -> List[str]: ... # pragma: no cover @runtime_checkable @@ -400,8 +397,7 @@ def __call__( endidx: int, *, arg_tokens: Dict[str, List[str]] = {}, - ) -> List[str]: - ... # pragma: no cover + ) -> List[str]: ... # pragma: no cover CompleterFunc = Union[CompleterFuncBase, CompleterFuncWithTokens] diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index 5344b5ab..e0f2259d 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -854,7 +854,7 @@ def _register_subcommands(self, cmdset: Union[CommandSet, 'Cmd']) -> None: ) # iterate through all matching methods - for method_name, method in methods: + for _, method in methods: subcommand_name: str = getattr(method, constants.SUBCMD_ATTR_NAME) full_command_name: str = getattr(method, constants.SUBCMD_ATTR_COMMAND) subcmd_parser_builder = getattr(method, constants.CMD_ATTR_ARGPARSER) @@ -963,7 +963,7 @@ def _unregister_subcommands(self, cmdset: Union[CommandSet, 'Cmd']) -> None: ) # iterate through all matching methods - for method_name, method in methods: + for _, method in methods: subcommand_name = getattr(method, constants.SUBCMD_ATTR_NAME) command_name = getattr(method, constants.SUBCMD_ATTR_COMMAND) @@ -2430,29 +2430,22 @@ def sigint_handler(self, signum: int, _: FrameType) -> None: if raise_interrupt: self._raise_keyboard_interrupt() - def sigterm_handler(self, signum: int, _: FrameType) -> None: # pragma: no cover + def termination_signal_handler(self, signum: int, _: FrameType) -> None: """ - Signal handler for SIGTERMs which are sent to politely ask this app to terminate. + Signal handler for SIGHUP and SIGTERM. Only runs on Linux and Mac. - If you need custom SIGTERM behavior, then override this method. + SIGHUP - received when terminal window is closed. + SIGTERM - received when this process politely asked to terminate. - :param signum: signal number - :param _: required param for signal handlers - """ - # Gracefully exit so the persistent history file will be written. - sys.exit(self.exit_code) - - def sighup_handler(self, signum: int, _: FrameType) -> None: # pragma: no cover - """ - Signal handler for SIGHUPs which are sent when the terminal is closed. - - If you need custom SIGHUP behavior, then override this method. + The basic purpose of this method is to call sys.exit() so our atexit handler will run + and save the persistent history file. If you need more complex behavior like killing + threads and performing cleanup, then override this method. :param signum: signal number :param _: required param for signal handlers """ - # Gracefully exit so the persistent history file will be written. - sys.exit(self.exit_code) + # POSIX systems add 128 to signal numbers for the exit code + sys.exit(128 + signum) def _raise_keyboard_interrupt(self) -> None: """Helper function to raise a KeyboardInterrupt""" @@ -5458,17 +5451,18 @@ def cmdloop(self, intro: Optional[str] = None) -> int: # type: ignore[override] if not threading.current_thread() is threading.main_thread(): raise RuntimeError("cmdloop must be run in the main thread") - # Register a SIGINT signal handler for Ctrl+C + # Register signal handlers import signal original_sigint_handler = signal.getsignal(signal.SIGINT) signal.signal(signal.SIGINT, self.sigint_handler) # type: ignore - original_sigterm_handler = signal.getsignal(signal.SIGTERM) - signal.signal(signal.SIGTERM, self.sigterm_handler) # type: ignore + if not sys.platform.startswith('win'): + original_sighup_handler = signal.getsignal(signal.SIGHUP) + signal.signal(signal.SIGHUP, self.termination_signal_handler) # type: ignore - original_sighup_handler = signal.getsignal(signal.SIGHUP) - signal.signal(signal.SIGHUP, self.sighup_handler) # type: ignore + original_sigterm_handler = signal.getsignal(signal.SIGTERM) + signal.signal(signal.SIGTERM, self.termination_signal_handler) # type: ignore # Grab terminal lock before the command line prompt has been drawn by readline self.terminal_lock.acquire() @@ -5502,10 +5496,12 @@ def cmdloop(self, intro: Optional[str] = None) -> int: # type: ignore[override] # This will also zero the lock count in case cmdloop() is called again self.terminal_lock.release() - # Restore the original signal handlers + # Restore original signal handlers signal.signal(signal.SIGINT, original_sigint_handler) - signal.signal(signal.SIGTERM, original_sigterm_handler) - signal.signal(signal.SIGHUP, original_sighup_handler) + + if not sys.platform.startswith('win'): + signal.signal(signal.SIGHUP, original_sighup_handler) + signal.signal(signal.SIGTERM, original_sigterm_handler) return self.exit_code diff --git a/cmd2/decorators.py b/cmd2/decorators.py index 3a163fda..361b71ac 100644 --- a/cmd2/decorators.py +++ b/cmd2/decorators.py @@ -266,8 +266,7 @@ def with_argparser( ns_provider: Optional[Callable[..., argparse.Namespace]] = None, preserve_quotes: bool = False, with_unknown_args: bool = False, -) -> Callable[[ArgparseCommandFunc[CommandParent]], RawCommandFuncOptionalBoolReturn[CommandParent]]: - ... # pragma: no cover +) -> Callable[[ArgparseCommandFunc[CommandParent]], RawCommandFuncOptionalBoolReturn[CommandParent]]: ... # pragma: no cover @overload @@ -277,8 +276,7 @@ def with_argparser( ns_provider: Optional[Callable[..., argparse.Namespace]] = None, preserve_quotes: bool = False, with_unknown_args: bool = False, -) -> Callable[[ArgparseCommandFunc[CommandParent]], RawCommandFuncOptionalBoolReturn[CommandParent]]: - ... # pragma: no cover +) -> Callable[[ArgparseCommandFunc[CommandParent]], RawCommandFuncOptionalBoolReturn[CommandParent]]: ... # pragma: no cover def with_argparser( @@ -418,8 +416,7 @@ def as_subcommand_to( *, help: Optional[str] = None, aliases: Optional[List[str]] = None, -) -> Callable[[ArgparseCommandFunc[CommandParent]], ArgparseCommandFunc[CommandParent]]: - ... # pragma: no cover +) -> Callable[[ArgparseCommandFunc[CommandParent]], ArgparseCommandFunc[CommandParent]]: ... # pragma: no cover @overload @@ -430,8 +427,7 @@ def as_subcommand_to( *, help: Optional[str] = None, aliases: Optional[List[str]] = None, -) -> Callable[[ArgparseCommandFunc[CommandParent]], ArgparseCommandFunc[CommandParent]]: - ... # pragma: no cover +) -> Callable[[ArgparseCommandFunc[CommandParent]], ArgparseCommandFunc[CommandParent]]: ... # pragma: no cover def as_subcommand_to( diff --git a/cmd2/history.py b/cmd2/history.py index a7d6baff..c79a19dd 100644 --- a/cmd2/history.py +++ b/cmd2/history.py @@ -154,12 +154,10 @@ def _zero_based_index(self, onebased: Union[int, str]) -> int: return result @overload - def append(self, new: HistoryItem) -> None: - ... # pragma: no cover + def append(self, new: HistoryItem) -> None: ... # pragma: no cover @overload - def append(self, new: Statement) -> None: - ... # pragma: no cover + def append(self, new: Statement) -> None: ... # pragma: no cover def append(self, new: Union[Statement, HistoryItem]) -> None: """Append a new statement to the end of the History list. diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py index 5686e9f9..2130cbf1 100755 --- a/tests/test_cmd2.py +++ b/tests/test_cmd2.py @@ -1038,6 +1038,17 @@ def test_raise_keyboard_interrupt(base_app): assert 'Got a keyboard interrupt' in str(excinfo.value) +@pytest.mark.skipif(sys.platform.startswith('win'), reason="SIGTERM only handeled on Linux/Mac") +def test_termination_signal_handler(base_app): + with pytest.raises(SystemExit) as excinfo: + base_app.termination_signal_handler(signal.SIGHUP, 1) + assert excinfo.value.code == signal.SIGHUP + 128 + + with pytest.raises(SystemExit) as excinfo: + base_app.termination_signal_handler(signal.SIGTERM, 1) + assert excinfo.value.code == signal.SIGTERM + 128 + + class HookFailureApp(cmd2.Cmd): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs)