[AppService] Fix #34005: az webapp ssh: Report SSH session failures instead of exiting 0

src/azure-cli/azure/cli/command_modules/appservice/custom.py

@@ -11401,8 +11401,15 @@ def create_tunnel_and_session(cmd, resource_group_name, name, port=None, slot=No
     ssh_user_name = 'root'
     ssh_user_password = 'Docker!'
 
-    s = threading.Thread(target=_start_ssh_session,
-                         args=('localhost', tunnel_server.get_port(), ssh_user_name, ssh_user_password))
+    ssh_exception_holder = [None]
+
+    def _ssh_session_with_error_capture():
+        try:
+            _start_ssh_session('localhost', tunnel_server.get_port(), ssh_user_name, ssh_user_password)
+        except Exception as ex:  # pylint: disable=broad-except
+            ssh_exception_holder[0] = ex
+
+    s = threading.Thread(target=_ssh_session_with_error_capture)
     s.daemon = True
     s.start()
 
@@ -11412,6 +11419,10 @@ def create_tunnel_and_session(cmd, resource_group_name, name, port=None, slot=No
         while s.is_alive() and t.is_alive():
             time.sleep(5)
 
+    ssh_ex = ssh_exception_holder[0]
+    if ssh_ex is not None:
+        raise ssh_ex
+
 
 def perform_onedeploy_functionapp(cmd,
                                   resource_group_name,
@@ -12125,7 +12136,7 @@ def _start_ssh_session(hostname, port, username, password):
             pass
         c.run('source /etc/profile; exec $SHELL -l', pty=True)
     except Exception as ex:  # pylint: disable=broad-except
-        logger.info(ex)
+        raise CLIError("SSH session failed: {}".format(ex)) from ex
     finally:
         c.close()
 

src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_webapp_commands_thru_mock.py

@@ -2560,5 +2560,97 @@ def test_get_java_runtimes_from_container_settings_reads_mapping(self):
         self.assertTrue(all(is_auto for _, _, is_auto in runtimes))
 
 
+class TestSshSessionFailureReporting(unittest.TestCase):
+    """Tests that SSH session failures are reported instead of silently exiting 0."""
+
+    @mock.patch('azure.cli.command_modules.appservice.custom.Connection')
+    def test_start_ssh_session_raises_clierror_on_run_failure(self, mock_connection_cls):
+        """_start_ssh_session raises CLIError when c.run() raises an exception."""
+        from knack.util import CLIError
+        from azure.cli.command_modules.appservice.custom import _start_ssh_session
+
+        mock_conn = mock.MagicMock()
+        mock_conn.run.side_effect = Exception("buffer overflow")
+        mock_connection_cls.return_value = mock_conn
+
+        with self.assertRaises(CLIError) as ctx:
+            _start_ssh_session('localhost', 2222, 'root', 'Docker!')
+
+        self.assertIn("SSH session failed", str(ctx.exception))
+        mock_conn.close.assert_called_once()
+
+    @mock.patch('azure.cli.command_modules.appservice.custom.Connection')
+    def test_start_ssh_session_closes_connection_on_failure(self, mock_connection_cls):
+        """_start_ssh_session always closes the connection even when an error occurs."""
+        from knack.util import CLIError
+        from azure.cli.command_modules.appservice.custom import _start_ssh_session
+
+        mock_conn = mock.MagicMock()
+        mock_conn.run.side_effect = RuntimeError("ioctl error")
+        mock_connection_cls.return_value = mock_conn
+
+        with self.assertRaises(CLIError):
+            _start_ssh_session('localhost', 2222, 'root', 'Docker!')
+
+        mock_conn.close.assert_called_once()
+
+    @mock.patch('azure.cli.command_modules.appservice.custom._start_tunnel')
+    @mock.patch('azure.cli.command_modules.appservice.custom.get_tunnel')
+    def test_create_tunnel_and_session_propagates_ssh_error(self, mock_get_tunnel, mock_start_tunnel):
+        """create_tunnel_and_session raises when the SSH session thread encounters an error."""
+        import threading
+        from knack.util import CLIError
+        from azure.cli.command_modules.appservice.custom import create_tunnel_and_session
+
+        mock_tunnel_server = mock.MagicMock()
+        mock_tunnel_server.get_port.return_value = 2222
+        mock_get_tunnel.return_value = mock_tunnel_server
+
+        ssh_error = CLIError("SSH session failed: buffer overflow")
+
+        def fake_ssh_session(hostname, port, username, password):
+            raise ssh_error
+
+        with mock.patch('azure.cli.command_modules.appservice.custom._start_ssh_session',
+                        side_effect=fake_ssh_session):
+            with mock.patch('azure.cli.command_modules.appservice.custom.threading.Thread') as mock_thread_cls:
+                # The tunnel thread (t) is created first; the SSH thread (s) is created second.
+                tunnel_thread = mock.MagicMock()
+                tunnel_thread.is_alive.return_value = False
+
+                ssh_thread = mock.MagicMock()
+                ssh_thread.is_alive.return_value = False
+
+                # Use a call counter to distinguish the two Thread() calls.
+                call_order = [0]
+                ssh_thread_target_holder = []
+
+                def capturing_thread_factory(*args, **kwargs):
+                    call_index = call_order[0]
+                    call_order[0] += 1
+                    # First call is the tunnel thread, second is the SSH thread.
+                    if call_index == 0:
+                        return tunnel_thread
+                    target = kwargs.get('target')
+                    if target is not None:
+                        ssh_thread_target_holder.append(target)
+                    return ssh_thread
+
+                mock_thread_cls.side_effect = capturing_thread_factory
+
+                def run_ssh_thread_synchronously():
+                    if ssh_thread_target_holder:
+                        ssh_thread_target_holder[0]()
+
+                ssh_thread.start.side_effect = run_ssh_thread_synchronously
+
+                cmd = mock.MagicMock()
+
+                with self.assertRaises(CLIError) as ctx:
+                    create_tunnel_and_session(cmd, 'rg', 'myapp', timeout=None)
+
+                self.assertIn("SSH session failed", str(ctx.exception))
+
+
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()