From e7a2e2283212642d3d58ff3e1afda7bf65922814 Mon Sep 17 00:00:00 2001 From: "pull-apart[bot]" <270479319+pull-apart[bot]@users.noreply.github.com> Date: Mon, 22 Jun 2026 10:57:10 +0000 Subject: [PATCH] split-pr: refactor bucket (from #21) --- utilities/unittests/test_console.py | 293 +++++++++++++++++++++++++--- 1 file changed, 265 insertions(+), 28 deletions(-) diff --git a/utilities/unittests/test_console.py b/utilities/unittests/test_console.py index ca8bae0396..e4ba5fe17e 100644 --- a/utilities/unittests/test_console.py +++ b/utilities/unittests/test_console.py @@ -3,11 +3,31 @@ """Unit tests for console module""" import os +import subprocess from unittest.mock import MagicMock, mock_open, patch +import pexpect +import pytest from console import Console +class SingleAttemptTimeoutSampler: + """Run a TimeoutSampler-wrapped function once without retries. + + Patch ``timeout_sampler.TimeoutSampler`` (not ``console.TimeoutSampler``): + ``connect()`` uses ``@retry`` from timeout_sampler, which resolves + TimeoutSampler from that module at runtime. ``console_eof_sampler`` uses + the separate ``console.TimeoutSampler`` import binding. + """ + + def __init__(self, func, func_args=(), **kwargs): + self.func = func + self.func_args = func_args + + def __iter__(self): + yield self.func(*self.func_args) + + class TestConsole: """Test cases for Console class""" @@ -313,10 +333,9 @@ def test_console_disconnect_no_username(self, mock_get_dir): exit_calls = [call for call in console.child.send.call_args_list if "exit" in str(call)] assert len(exit_calls) == 0 - @patch("console.pexpect") @patch("console.get_data_collector_base_directory") - def test_console_disconnect_terminated_child(self, mock_get_dir, mock_pexpect): - """Test disconnect method when child is terminated""" + def test_console_disconnect_terminated_child(self, mock_get_dir): + """Test disconnect method when the console subprocess has exited""" mock_get_dir.return_value = "/tmp/data" mock_vm = MagicMock() mock_vm.name = "test-vm" @@ -327,19 +346,14 @@ def test_console_disconnect_terminated_child(self, mock_get_dir, mock_pexpect): console = Console(vm=mock_vm) console.child = MagicMock() - console.child.terminated = True - - # Mock the console_eof_sampler - console.console_eof_sampler = MagicMock() + mock_proc = MagicMock() + mock_proc.poll.return_value = 0 + console._proc = mock_proc - console.disconnect() + with patch.object(console, "console_eof_sampler") as mock_eof_sampler: + console.disconnect() - # Should call console_eof_sampler when child is terminated - console.console_eof_sampler.assert_called_once_with( - func=mock_pexpect.spawn, - command=console.cmd, - timeout=console.timeout, - ) + mock_eof_sampler.assert_called_once_with() @patch("console.TimeoutSampler") @patch("builtins.open", new_callable=mock_open) @@ -363,18 +377,12 @@ def test_console_eof_sampler_success(self, mock_get_dir, mock_file_open, mock_ti mock_sampler_instance.__iter__.return_value = [mock_sample] mock_timeout_sampler.return_value = mock_sampler_instance - mock_func = MagicMock() - command = "test-command" - timeout = 30 + console.console_eof_sampler() - console.console_eof_sampler(func=mock_func, command=command, timeout=timeout) - - # Should create TimeoutSampler with correct parameters + # Should create TimeoutSampler with _spawn_console as the retry function mock_timeout_sampler.assert_called_once() call_args = mock_timeout_sampler.call_args - assert call_args[1]["func"] == mock_func - assert call_args[1]["command"] == command - assert call_args[1]["timeout"] == timeout + assert call_args[1]["func"] == console._spawn_console # Should set child and logfile assert console.child == mock_sample @@ -400,11 +408,240 @@ def test_console_eof_sampler_no_sample(self, mock_get_dir, mock_timeout_sampler) mock_sampler_instance.__iter__.return_value = [None] mock_timeout_sampler.return_value = mock_sampler_instance - mock_func = MagicMock() - command = "test-command" - timeout = 30 - - console.console_eof_sampler(func=mock_func, command=command, timeout=timeout) + console.console_eof_sampler() # Should not change child when no valid sample is found assert console.child == original_child + + @patch("console.os.close") + @patch("console.pexpect.fdpexpect.fdspawn") + @patch("console.subprocess.Popen") + @patch("console.pty.openpty") + @patch("console.get_data_collector_base_directory") + def test_spawn_console_success( + self, + mock_get_dir, + mock_openpty, + mock_popen, + mock_fdspawn, + mock_os_close, + ): + """Test _spawn_console creates fdspawn and stores subprocess handle""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_openpty.return_value = (10, 11) + mock_proc = MagicMock() + mock_popen.return_value = mock_proc + mock_child = MagicMock() + mock_fdspawn.return_value = mock_child + + console = Console(vm=mock_vm) + result = console._spawn_console() + + assert result == mock_child + assert console._proc == mock_proc + mock_popen.assert_called_once() + mock_fdspawn.assert_called_once_with(fd=10, encoding="utf-8", timeout=30) + mock_os_close.assert_called_once_with(11) + + @patch("console.os.close") + @patch("console.subprocess.Popen") + @patch("console.pty.openpty") + @patch("console.get_data_collector_base_directory") + def test_spawn_console_popen_failure(self, mock_get_dir, mock_openpty, mock_popen, mock_os_close): + """Test _spawn_console closes pty fds when subprocess spawn fails""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_openpty.return_value = (10, 11) + mock_popen.side_effect = OSError("spawn failed") + + console = Console(vm=mock_vm) + with pytest.raises(OSError, match="spawn failed"): + console._spawn_console() + + mock_os_close.assert_any_call(10) + mock_os_close.assert_any_call(11) + assert console._proc is None + + @patch("console.os.close") + @patch("console.pexpect.fdpexpect.fdspawn") + @patch("console.subprocess.Popen") + @patch("console.pty.openpty") + @patch("console.get_data_collector_base_directory") + def test_spawn_console_fdspawn_failure( + self, + mock_get_dir, + mock_openpty, + mock_popen, + mock_fdspawn, + mock_os_close, + ): + """Test _spawn_console terminates subprocess when fdspawn fails""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_openpty.return_value = (10, 11) + mock_proc = MagicMock() + mock_popen.return_value = mock_proc + mock_fdspawn.side_effect = pexpect.exceptions.ExceptionPexpect("fdspawn failed") + + console = Console(vm=mock_vm) + with pytest.raises(pexpect.exceptions.ExceptionPexpect, match="fdspawn failed"): + console._spawn_console() + + mock_proc.terminate.assert_called_once() + mock_proc.wait.assert_called_once() + mock_os_close.assert_any_call(10) + mock_os_close.assert_any_call(11) + assert console._proc is None + + @patch("console.os.close") + @patch("console.pexpect.fdpexpect.fdspawn") + @patch("console.subprocess.Popen") + @patch("console.pty.openpty") + @patch("console.get_data_collector_base_directory") + def test_spawn_console_fdspawn_failure_force_kill( + self, + mock_get_dir, + mock_openpty, + mock_popen, + mock_fdspawn, + mock_os_close, + ): + """Test _spawn_console force-kills subprocess when terminate times out""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_openpty.return_value = (10, 11) + mock_proc = MagicMock() + mock_popen.return_value = mock_proc + mock_proc.wait.side_effect = [subprocess.TimeoutExpired(cmd="virtctl", timeout=10), None] + mock_fdspawn.side_effect = pexpect.exceptions.ExceptionPexpect("fdspawn failed") + + console = Console(vm=mock_vm) + with pytest.raises(pexpect.exceptions.ExceptionPexpect, match="fdspawn failed"): + console._spawn_console() + + mock_proc.terminate.assert_called_once() + mock_proc.kill.assert_called_once() + assert mock_proc.wait.call_count == 2 + + @patch("console.get_data_collector_base_directory") + def test_terminate_proc_running_process(self, mock_get_dir): + """Test _terminate_proc gracefully terminates a running subprocess""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_proc = MagicMock() + mock_proc.poll.return_value = None + + console = Console(vm=mock_vm) + console._proc = mock_proc + console._terminate_proc() + + mock_proc.terminate.assert_called_once() + mock_proc.wait.assert_called_once_with(timeout=10) + assert console._proc is None + + @patch("console.get_data_collector_base_directory") + def test_terminate_proc_force_kill(self, mock_get_dir): + """Test _terminate_proc force-kills subprocess when terminate times out""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_proc = MagicMock() + mock_proc.poll.return_value = None + mock_proc.wait.side_effect = [subprocess.TimeoutExpired(cmd="virtctl", timeout=10), None] + + console = Console(vm=mock_vm) + console._proc = mock_proc + console._terminate_proc() + + mock_proc.terminate.assert_called_once() + mock_proc.kill.assert_called_once() + assert mock_proc.wait.call_count == 2 + assert console._proc is None + + @patch("console.get_data_collector_base_directory") + def test_terminate_proc_already_exited(self, mock_get_dir): + """Test _terminate_proc waits without terminate when process already exited""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + + mock_proc = MagicMock() + mock_proc.poll.return_value = 0 + + console = Console(vm=mock_vm) + console._proc = mock_proc + console._terminate_proc() + + mock_proc.terminate.assert_not_called() + mock_proc.wait.assert_called_once_with(timeout=10) + assert console._proc is None + + @patch("console.get_data_collector_base_directory") + def test_console_connect_failure_cleanup(self, mock_get_dir): + """Test connect cleans up child and subprocess on failure""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + mock_vm.username = "user" + mock_vm.password = "pass" + mock_vm.login_params = {} + + console = Console(vm=mock_vm) + mock_child = MagicMock() + + with ( + patch("timeout_sampler.TimeoutSampler", SingleAttemptTimeoutSampler), + patch.object(console, "console_eof_sampler") as mock_sampler, + patch.object(console, "_connect", side_effect=RuntimeError("connect failed")), + patch.object(console, "_terminate_proc") as mock_terminate, + ): + mock_sampler.side_effect = lambda: setattr(console, "child", mock_child) + with pytest.raises(RuntimeError, match="connect failed"): + console.connect() + + mock_child.close.assert_called_once() + mock_terminate.assert_called_once() + + @patch("console.get_data_collector_base_directory") + def test_console_connect_failure_no_child(self, mock_get_dir): + """Test connect terminates subprocess when child was never set""" + mock_get_dir.return_value = "/tmp/data" + mock_vm = MagicMock() + mock_vm.name = "test-vm" + mock_vm.namespace = None + mock_vm.username = "user" + mock_vm.password = "pass" + mock_vm.login_params = {} + + console = Console(vm=mock_vm) + + with ( + patch("timeout_sampler.TimeoutSampler", SingleAttemptTimeoutSampler), + patch.object(console, "console_eof_sampler"), + patch.object(console, "_connect", side_effect=RuntimeError("connect failed")), + patch.object(console, "_terminate_proc") as mock_terminate, + ): + with pytest.raises(RuntimeError, match="connect failed"): + console.connect() + + mock_terminate.assert_called_once()