From 1feaf824097c9d2383dfb0463e8e44efc3fec218 Mon Sep 17 00:00:00 2001 From: DawidBerk Date: Fri, 28 Aug 2026 10:32:28 +0200 Subject: [PATCH] feat: Add EOF and TIMEOUT exception handling in SolConnection._deactivate_sol_session Signed-off-by: Berk, Dawid --- mfd_connect/sol.py | 122 ++++++++- tests/unit/test_mfd_connect/test_sol.py | 326 ++++++++++++++++++++++++ 2 files changed, 442 insertions(+), 6 deletions(-) diff --git a/mfd_connect/sol.py b/mfd_connect/sol.py index a573e54..ce554c2 100644 --- a/mfd_connect/sol.py +++ b/mfd_connect/sol.py @@ -5,6 +5,7 @@ import logging import platform import re +import signal import sys import time import typing @@ -327,17 +328,126 @@ def _establish_connection(self, retry_count: int = 0) -> "pexpect.spawn": return sol_connection_process - def _deactivate_sol_session(self) -> None: - # clearing old session + def _deactivate_sol_session(self, retry_count: int = 2) -> None: + """ + Clear/deactivate previous SoL IPMI session before opening a new one. + + Retries graceful deactivation via ipmiutil, then falls back to killing + defunct ipmiutil process(es) if all attempts fail. + + :param retry_count: Number of graceful deactivation attempts + :raises SolException: if graceful deactivation and fallback cleanup fail + """ logger.log(level=log_levels.MODULE_DEBUG, msg="Clearing/Deactivating previous SoL IPMI session...") - process = pexpect.popen_spawn.PopenSpawn(f"{self._ipmi_tool_name} sol -d {self._ipmi_parameters}", timeout=60) + # Both patterns below signal a successful deactivation (either the session was closed, + # or there was none to close in the first place) - pexpect.expect() raises TIMEOUT/EOF + # if neither is matched, so a returned index always means success. correct_responses = [ "completed successfully", "Invalid Session Handle or Empty Buffer", ] - expect_index = process.expect(correct_responses) - if expect_index > len(correct_responses) - 1: - raise SolException(f"Fatal Error while deactivating previous SoL session! \n{process.before}") + + for attempt in range(1, retry_count + 1): + logger.log( + level=log_levels.MODULE_DEBUG, + msg=f"Attempt {attempt}/{retry_count} to deactivate previous SoL session...", + ) + process = None + try: + process = pexpect.popen_spawn.PopenSpawn( + f"{self._ipmi_tool_name} sol -d {self._ipmi_parameters}", timeout=60 + ) + process.expect(correct_responses) + except (OSError, pexpect.TIMEOUT, pexpect.EOF) as e: + logger.log( + level=log_levels.MODULE_DEBUG, + msg=f"Attempt {attempt}/{retry_count}: " + f"{self._ipmi_tool_name} did not respond as expected while deactivating SoL session: {e}", + ) + if process is not None: + try: + # PopenSpawn has no close() - kill() sends a signal via the underlying subprocess. + process.kill(getattr(signal, "SIGKILL", signal.SIGTERM)) + except (ProcessLookupError, OSError): + # Process already gone, or killing it failed for another OS-level reason (e.g. + # permissions) - not fatal here, the fallback cleanup below will handle it. + pass + continue + + logger.log(level=log_levels.MODULE_DEBUG, msg="...Done - previous SoL session deactivated.") + return + + logger.log( + level=log_levels.MODULE_DEBUG, + msg=f"Could not gracefully deactivate SoL session after {retry_count} attempt(s). " + f"Trying to locate and kill defunct {self._ipmi_tool_name} process(es)...", + ) + self._kill_defunct_ipmiutil_processes() + + def _kill_defunct_ipmiutil_processes(self) -> None: + """ + Find and kill defunct/stopped ipmiutil process(es). + + Process listing uses 'ps -o pid,stat --no-headers -C ': '-C ' + selects processes by command name and, on its own, already overrides the default + "only same-terminal processes" restriction of 'ps', matching system-wide. + + Only processes in STAT='T' (stopped by job control signal) are considered defunct + and killed, healthy ipmiutil is usually 'S' or 'S+' (interruptible sleep). + + Finding no defunct process is not treated as fatal, it simply means there's + nothing to clean up. + + :raises SolException: if process listing fails or kill operation fails + """ + try: + ps_process = pexpect.popen_spawn.PopenSpawn( + f"ps -o pid,stat --no-headers -C {self._ipmi_tool_name}", timeout=30 + ) + # Accept TIMEOUT as well as EOF: if 'ps' hasn't finished within 30s, use whatever + # partial output was captured in ps_process.before instead of raising an exception. + ps_process.expect([pexpect.EOF, pexpect.TIMEOUT]) + ps_output = ps_process.before.decode("ASCII", errors="ignore") + except (OSError, pexpect.ExceptionPexpect) as e: + raise SolException(f"Fatal Error while searching for defunct {self._ipmi_tool_name} process(es)! \n{e}") + + defunct_pids = [] + for line in ps_output.splitlines(): + columns = line.split() + if len(columns) < 2: + continue + pid, stat = columns[0], columns[1] + if stat.startswith("T"): + defunct_pids.append(pid) + else: + logger.log( + level=log_levels.MODULE_DEBUG, + msg=f"{self._ipmi_tool_name} process PID {pid} is in a healthy state '{stat}' " + "- leaving it running.", + ) + + if not defunct_pids: + logger.log( + level=log_levels.MODULE_DEBUG, + msg=f"No defunct (stopped, STAT=T) {self._ipmi_tool_name} process(es) found - nothing to clean up.", + ) + return + + logger.log( + level=log_levels.MODULE_DEBUG, + msg=f"Found defunct {self._ipmi_tool_name} process(es) with PID(s): {defunct_pids}. Killing them...", + ) + for pid in defunct_pids: + logger.log(level=log_levels.MODULE_DEBUG, msg=f"Killing defunct {self._ipmi_tool_name} process, PID: {pid}") + try: + # sudo -n (non-interactive) fails immediately if a password prompt would appear, + # avoiding indefinite blocking in automation. + kill_process = pexpect.popen_spawn.PopenSpawn(f"sudo -n kill -KILL {pid}", timeout=30) + kill_process.expect(pexpect.EOF) + except (OSError, pexpect.ExceptionPexpect) as e: + raise SolException(f"Fatal Error while killing defunct {self._ipmi_tool_name} process PID {pid}! \n{e}") + + logger.log(level=log_levels.MODULE_DEBUG, msg=f"...Done - defunct {self._ipmi_tool_name} process(es) killed.") @log_func_info(logger) def wait_for_string(self, string_list: List[str], expect_timeout: bool = False, timeout: int = 30) -> int: diff --git a/tests/unit/test_mfd_connect/test_sol.py b/tests/unit/test_mfd_connect/test_sol.py index 21e8c09..5e1fe53 100644 --- a/tests/unit/test_mfd_connect/test_sol.py +++ b/tests/unit/test_mfd_connect/test_sol.py @@ -1,9 +1,12 @@ # Copyright (C) 2025-2026 Intel Corporation # SPDX-License-Identifier: MIT +import signal import sys from subprocess import CalledProcessError from textwrap import dedent +import pexpect +import pexpect.popen_spawn import pytest from mfd_typing.os_values import OSBitness, OSType, OSName from pytest import raises, fixture @@ -17,6 +20,8 @@ class TestSolConnection: """Tests of SolConnection.""" + KILL_SIGNAL = getattr(signal, "SIGKILL", signal.SIGTERM) + CustomTestException = CalledProcessError @fixture @@ -152,6 +157,327 @@ def test_establish_connection_retries_once(self, sol, mocker): assert spawn.call_count == 2 assert result is second_child + def test_deactivate_sol_session_success_first_attempt(self, sol, mocker): + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + child = mocker.Mock() + child.expect.return_value = 0 + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", return_value=child) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() + + popen_spawn.assert_called_once() + assert popen_spawn.call_args.args[0] == "ipmiutil sol -d -F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + kill_mock.assert_not_called() + + def test_deactivate_sol_session_default_retry_count_is_two(self, sol, mocker): + """Task requirement: exactly 2 graceful deactivation attempts before falling back to killing.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + child.expect.side_effect = pexpect.TIMEOUT("timed out") + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", return_value=child) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() # no retry_count passed - must default to 2 + + assert popen_spawn.call_count == 2 + kill_mock.assert_called_once() + + def test_deactivate_sol_session_handles_oserror_from_popen_spawn(self, sol, mocker): + """If ipmiutil is missing, PopenSpawn itself raises OSError - must be tolerated like TIMEOUT/EOF.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", + side_effect=OSError("ipmiutil: command not found"), + ) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() # must not raise + + assert popen_spawn.call_count == 2 + kill_mock.assert_called_once() + + def test_deactivate_sol_session_retries_after_timeout_then_succeeds(self, sol, mocker): + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + first_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + first_child.expect.side_effect = pexpect.TIMEOUT("timed out") + second_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + second_child.expect.return_value = 0 + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[first_child, second_child] + ) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() + + assert popen_spawn.call_count == 2 + first_child.kill.assert_called_once_with(self.KILL_SIGNAL) + second_child.kill.assert_not_called() + kill_mock.assert_not_called() + + def test_deactivate_sol_session_falls_back_to_kill_after_repeated_failures(self, sol, mocker): + """Both attempts raise pexpect exceptions (defunct process) - the whole run must not crash.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + first_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + first_child.expect.side_effect = pexpect.TIMEOUT("timed out") + second_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + second_child.expect.side_effect = pexpect.EOF("eof") + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[first_child, second_child] + ) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() # must not raise + + assert popen_spawn.call_count == 2 + first_child.kill.assert_called_once_with(self.KILL_SIGNAL) + second_child.kill.assert_called_once_with(self.KILL_SIGNAL) + kill_mock.assert_called_once() + + def test_deactivate_sol_session_ignores_process_already_gone(self, sol, mocker): + """If the lingering process already exited (ProcessLookupError on kill), it must not crash the retry.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + first_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + first_child.expect.side_effect = pexpect.EOF("eof") + first_child.kill.side_effect = ProcessLookupError("no such process") + second_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + second_child.expect.return_value = 0 + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[first_child, second_child] + ) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() # must not raise despite ProcessLookupError + + assert popen_spawn.call_count == 2 + kill_mock.assert_not_called() + + def test_deactivate_sol_session_ignores_oserror_on_lingering_process_kill(self, sol, mocker): + """If killing the lingering process fails with a generic OSError (e.g. permissions), retry must continue.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + first_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + first_child.expect.side_effect = pexpect.TIMEOUT("timed out") + first_child.kill.side_effect = OSError("permission denied") + second_child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + second_child.expect.return_value = 0 + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[first_child, second_child] + ) + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session() # must not raise despite OSError from kill() + + assert popen_spawn.call_count == 2 + kill_mock.assert_not_called() + + def test_deactivate_sol_session_propagates_kill_failure(self, sol, mocker): + """If the fallback kill of defunct processes fails, its SolException must propagate, not be swallowed.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + child = mocker.create_autospec(pexpect.popen_spawn.PopenSpawn, instance=True) + child.expect.side_effect = pexpect.TIMEOUT("timed out") + mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", return_value=child) + mocker.patch.object( + sol, "_kill_defunct_ipmiutil_processes", side_effect=SolException("no defunct process found") + ) + + with pytest.raises(SolException): + sol._deactivate_sol_session() + + def test_deactivate_sol_session_zero_retry_count_skips_graceful_attempts(self, sol, mocker): + """With retry_count=0 no graceful attempt is made at all - fallback kill triggers immediately.""" + sol._ipmi_tool_name = "ipmiutil" + sol._ipmi_parameters = "-F lan2 -U admin -P secret -N 10.10.10.10 -V 4" + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn") + kill_mock = mocker.patch.object(sol, "_kill_defunct_ipmiutil_processes") + + sol._deactivate_sol_session(retry_count=0) + + popen_spawn.assert_not_called() + kill_mock.assert_called_once() + + def test_kill_defunct_ipmiutil_processes_kills_found_pid(self, sol, mocker): + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 206865 T\n" + kill_child = mocker.Mock() + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + sol._kill_defunct_ipmiutil_processes() + + assert popen_spawn.call_args_list[0].args[0] == "ps -o pid,stat --no-headers -C ipmiutil" + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 206865" + ps_child.expect.assert_called_once_with([pexpect.EOF, pexpect.TIMEOUT]) + kill_child.expect.assert_called_once_with(pexpect.EOF) + + def test_kill_defunct_ipmiutil_processes_parses_real_ps_output_format(self, sol, mocker): + """Uses the exact real-world sample provided by reviewer: right-justified PID, multi-char STAT token.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 1094 Tl\n" + kill_child = mocker.Mock() + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + sol._kill_defunct_ipmiutil_processes() + + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 1094" + + def test_kill_defunct_ipmiutil_processes_kills_multiple_pids(self, sol, mocker): + """Mirrors real-world output: two stopped (T) ipmiutil sessions must both be killed.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 206865 T\n 210671 T\n" + kill_child_1 = mocker.Mock() + kill_child_2 = mocker.Mock() + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", + side_effect=[ps_child, kill_child_1, kill_child_2], + ) + + sol._kill_defunct_ipmiutil_processes() + + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 206865" + assert popen_spawn.call_args_list[2].args[0] == "sudo -n kill -KILL 210671" + + def test_kill_defunct_ipmiutil_processes_kills_only_stopped_ones(self, sol, mocker): + """Mix of stopped (T) and healthy (Sl, per real sample output) ipmiutil processes - only stopped is killed.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 206865 T\n 220000 Sl\n" + kill_child = mocker.Mock() + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + sol._kill_defunct_ipmiutil_processes() + + assert popen_spawn.call_count == 2 # ps listing + single kill (healthy one skipped) + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 206865" + + def test_kill_defunct_ipmiutil_processes_skips_malformed_lines(self, sol, mocker): + """Blank/incomplete lines (fewer than 2 columns) must be skipped instead of raising.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b"\n 206865\n 206865 T\n" + kill_child = mocker.Mock() + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + sol._kill_defunct_ipmiutil_processes() + + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 206865" + + def test_kill_defunct_ipmiutil_processes_accepts_ps_timeout_with_partial_output(self, sol, mocker): + """A TIMEOUT while listing processes is tolerated - partial output already captured is still used.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.expect.return_value = 1 # simulates pexpect.TIMEOUT being matched instead of pexpect.EOF + ps_child.before = b" 206865 T\n" + kill_child = mocker.Mock() + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + sol._kill_defunct_ipmiutil_processes() # must not raise despite the timeout + + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 206865" + + def test_kill_defunct_ipmiutil_processes_no_pid_found_returns_without_raising(self, sol, mocker): + """When no defunct (STAT=T) process is found, this is not fatal - the method must return normally.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 1094 Sl\n" + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", return_value=ps_child) + + sol._kill_defunct_ipmiutil_processes() # must not raise + + popen_spawn.assert_called_once() # only the ps listing call - no kill attempted + + def test_kill_defunct_ipmiutil_processes_no_match_at_all_returns_without_raising(self, sol, mocker): + """When 'ps -C' finds no matching process at all, output is empty - must also return normally.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b"" # 'ps -C ipmiutil' with no matching process produces empty output + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", return_value=ps_child) + + sol._kill_defunct_ipmiutil_processes() # must not raise + + popen_spawn.assert_called_once() + + def test_kill_defunct_ipmiutil_processes_raises_when_ps_listing_fails(self, sol, mocker): + """Any unexpected pexpect exception while listing processes must be wrapped into SolException.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.expect.side_effect = pexpect.TIMEOUT("timed out") + mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", return_value=ps_child) + + with pytest.raises(SolException): + sol._kill_defunct_ipmiutil_processes() + + def test_kill_defunct_ipmiutil_processes_raises_when_ps_spawn_fails_with_oserror(self, sol, mocker): + """OSError (e.g. FileNotFoundError if 'ps' is missing) while spawning must be wrapped into SolException.""" + sol._ipmi_tool_name = "ipmiutil" + mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=OSError("ps: command not found")) + + with pytest.raises(SolException): + sol._kill_defunct_ipmiutil_processes() + + def test_kill_defunct_ipmiutil_processes_raises_when_kill_spawn_fails_with_oserror(self, sol, mocker): + """OSError (e.g. FileNotFoundError if 'sudo' is missing) while spawning kill must be wrapped too.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 206865 T\n" + popen_spawn = mocker.patch( + "mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", + side_effect=[ps_child, OSError("sudo: command not found")], + ) + + with pytest.raises(SolException): + sol._kill_defunct_ipmiutil_processes() + + assert popen_spawn.call_count == 2 + + def test_kill_defunct_ipmiutil_processes_raises_when_kill_fails(self, sol, mocker): + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = b" 206865 T\n" + kill_child = mocker.Mock() + kill_child.expect.side_effect = pexpect.TIMEOUT("timed out") + mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + with pytest.raises(SolException): + sol._kill_defunct_ipmiutil_processes() + + @pytest.mark.parametrize( + "stat, should_be_killed", + [ + ("S", False), + ("S+", False), + ("Sl", False), + ("R", False), + ("R+", False), + ("T", True), + ("Tl", True), + ("T+", True), + ], + ) + def test_kill_defunct_ipmiutil_processes_stat_classification(self, sol, mocker, stat, should_be_killed): + """Only STAT starting with 'T' (stopped by job control signal) is treated as defunct, per task assumptions.""" + sol._ipmi_tool_name = "ipmiutil" + ps_child = mocker.Mock() + ps_child.before = f" 123456 {stat}\n".encode("ASCII") + kill_child = mocker.Mock() + popen_spawn = mocker.patch("mfd_connect.sol.pexpect.popen_spawn.PopenSpawn", side_effect=[ps_child, kill_child]) + + sol._kill_defunct_ipmiutil_processes() + + expected_call_count = 2 if should_be_killed else 1 + assert popen_spawn.call_count == expected_call_count + if should_be_killed: + assert popen_spawn.call_args_list[1].args[0] == "sudo -n kill -KILL 123456" + def test__parse_selection_regex_fallback_blue_background(self): output = "\x1b[44mSelected Boot Option"