From a4563db77a84b1c589e0e3f93059b46725d3a032 Mon Sep 17 00:00:00 2001 From: yoggydev <280342032+yoggydev@users.noreply.github.com> Date: Thu, 20 Aug 2026 17:40:52 +0900 Subject: [PATCH] test(cli): capture kill calls without mutable accumulators --- .../client/cli/commands/autoroute/process.py | 4 +-- .../client/cli/autoroute/test_process.py | 35 +++++++++++-------- 2 files changed, 22 insertions(+), 17 deletions(-) diff --git a/litellm/proxy/client/cli/commands/autoroute/process.py b/litellm/proxy/client/cli/commands/autoroute/process.py index ada55c92673..3c9d32ce4fb 100644 --- a/litellm/proxy/client/cli/commands/autoroute/process.py +++ b/litellm/proxy/client/cli/commands/autoroute/process.py @@ -159,7 +159,7 @@ def _windows_pid_exists(pid: int) -> bool: import ctypes from ctypes import wintypes - kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32: Final = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) kernel32.OpenProcess.restype = wintypes.HANDLE kernel32.GetExitCodeProcess.argtypes = (wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)) @@ -171,7 +171,7 @@ def _windows_pid_exists(pid: int) -> bool: if not handle: return ctypes.get_last_error() == _ERROR_ACCESS_DENIED try: - exit_code = wintypes.DWORD() + exit_code: Final = wintypes.DWORD() if not kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)): return True return exit_code.value == _STILL_ACTIVE diff --git a/tests/test_litellm/proxy/client/cli/autoroute/test_process.py b/tests/test_litellm/proxy/client/cli/autoroute/test_process.py index 51e020b450f..b0f1ba96636 100644 --- a/tests/test_litellm/proxy/client/cli/autoroute/test_process.py +++ b/tests/test_litellm/proxy/client/cli/autoroute/test_process.py @@ -4,8 +4,8 @@ import socket import subprocess import sys import time -from typing import Optional -from unittest.mock import patch +from typing import Final, Optional +from unittest.mock import Mock, patch import pytest @@ -39,6 +39,11 @@ class FakeResponse: self.status_code = status_code +def _signals_sent(fake_kill: Mock) -> tuple[int, ...]: + """Signal numbers ``os.kill`` was called with, in call order.""" + return tuple(call.args[1] for call in fake_kill.call_args_list) + + class TestIsPortAvailable: def test_true_for_a_free_port(self): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: @@ -147,7 +152,7 @@ class TestIsRunningDoesNotSignal: def test_probing_a_child_does_not_terminate_it(self): """The behavioural check, with nothing mocked.""" - child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + child: Final = subprocess.Popen((sys.executable, "-c", "import time; time.sleep(30)")) try: assert is_running(child.pid) is True time.sleep(1.0) @@ -171,13 +176,13 @@ class TestIsRunningDoesNotSignal: assert is_running(os.getpid()) is True def test_win32_answers_without_calling_os_kill(self, monkeypatch): - calls: list = [] + fake_kill: Final = Mock() monkeypatch.setattr(process_module.sys, "platform", "win32") - monkeypatch.setattr(process_module.os, "kill", lambda pid, sig: calls.append(sig)) + monkeypatch.setattr(process_module.os, "kill", fake_kill) monkeypatch.setattr(process_module, "_windows_pid_exists", lambda pid: True) assert is_running(1234) is True - assert calls == [], "os.kill reached on win32; signal 0 is CTRL_C_EVENT there" + assert _signals_sent(fake_kill) == (), "os.kill reached on win32; signal 0 is CTRL_C_EVENT there" class TestTerminateHardKillSignal: @@ -191,38 +196,38 @@ class TestTerminateHardKillSignal: than 0 or 1 routes to ``TerminateProcess`` on Windows, so SIGTERM is a real kill there. """ - sent: list = [] - monkeypatch.setattr(process_module.os, "kill", lambda pid, sig: sent.append(sig)) + fake_kill: Final = Mock() + monkeypatch.setattr(process_module.os, "kill", fake_kill) monkeypatch.setattr(process_module, "is_running", lambda pid: True) monkeypatch.setattr(process_module.time, "sleep", lambda seconds: None) monkeypatch.delattr(process_module.signal, "SIGKILL", raising=False) terminate(4242, grace_period=0.0) - assert sent == [signal.SIGTERM, signal.SIGTERM] + assert _signals_sent(fake_kill) == (signal.SIGTERM, signal.SIGTERM) @pytest.mark.skipif( not hasattr(signal, "SIGKILL"), reason="POSIX escalation path; SIGKILL is absent on this platform", ) def test_escalation_still_uses_sigkill_where_it_exists(self, monkeypatch): - sent: list = [] - monkeypatch.setattr(process_module.os, "kill", lambda pid, sig: sent.append(sig)) + fake_kill: Final = Mock() + monkeypatch.setattr(process_module.os, "kill", fake_kill) monkeypatch.setattr(process_module, "is_running", lambda pid: True) monkeypatch.setattr(process_module.time, "sleep", lambda seconds: None) terminate(4242, grace_period=0.0) - assert sent == [signal.SIGTERM, signal.SIGKILL] + assert _signals_sent(fake_kill) == (signal.SIGTERM, signal.SIGKILL) @pytest.mark.parametrize( "raised", - [ + ( PermissionError(5, "Access is denied"), OSError(22, "Invalid argument"), ProcessLookupError(3, "No such process"), - ], - ids=["permission-denied", "oserror", "already-gone"], + ), + ids=("permission-denied", "oserror", "already-gone"), ) def test_a_kill_that_cannot_land_does_not_escape(self, monkeypatch, raised): """A process that exited between the probe and the kill must not crash ``down``.