[split] Refactor / cleanup — from #21

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()