Fix IndexError in apache_beam.utils.processes when pip subprocess fails with short command

CHANGES.md

@@ -83,6 +83,7 @@
 
 ## Bugfixes
 
+* Fixed IndexError in `apache_beam.utils.processes` when pip subprocess fails with short command (e.g. `pip install pkg`) (Python) ([#37515](https://github.com/apache/beam/issues/37515)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes

sdks/python/apache_beam/utils/multi_process_shared_test.py

@@ -24,6 +24,8 @@
 import unittest
 from typing import Any
 
+import pytest
+
 from apache_beam.utils import multi_process_shared
 
 
@@ -85,6 +87,7 @@ def __getattribute__(self, __name: str) -> Any:
       return object.__getattribute__(self, __name)
 
 
+@pytest.mark.no_xdist
 class MultiProcessSharedTest(unittest.TestCase):
   @classmethod
   def setUpClass(cls):

sdks/python/apache_beam/utils/processes.py

@@ -44,6 +44,19 @@
 
 else:
 
+  def _pip_package_from_args(args):
+    """Return a safe string for the package field in pip error messages.
+
+    Avoids IndexError when the command list is shorter than 7 elements
+    (e.g. ['python', '-m', 'pip', 'install', 'pkg']).
+    """
+    if not isinstance(args, tuple) or not args:
+      return "see output below"
+    cmd = args[0]
+    if not isinstance(cmd, (list, tuple)) or len(cmd) <= 6:
+      return "see output below"
+    return cmd[6]
+
   def call(*args, **kwargs):
     if force_shell:
       kwargs['shell'] = True
@@ -52,11 +65,12 @@ def call(*args, **kwargs):
     except OSError as e:
       raise RuntimeError("Executable {} not found".format(args[0])) from e
     except subprocess.CalledProcessError as error:
-      if isinstance(args, tuple) and (args[0][2] == "pip"):
+      if isinstance(args, tuple) and len(args[0]) > 2 and args[0][2] == "pip":
         raise RuntimeError( \
           "Full traceback: {}\n Pip install failed for package: {} \
           \n Output from execution of subprocess: {}" \
-          .format(traceback.format_exc(), args[0][6], error. output)) from error
+          .format(traceback.format_exc(),
+                  _pip_package_from_args(args), error.output)) from error
       else:
         raise RuntimeError("Full trace: {}\
            \n Output of the failed child process: {} " \
@@ -71,11 +85,12 @@ def check_call(*args, **kwargs):
     except OSError as e:
       raise RuntimeError("Executable {} not found".format(args[0])) from e
     except subprocess.CalledProcessError as error:
-      if isinstance(args, tuple) and (args[0][2] == "pip"):
+      if isinstance(args, tuple) and len(args[0]) > 2 and args[0][2] == "pip":
         raise RuntimeError( \
           "Full traceback: {} \n Pip install failed for package: {} \
           \n Output from execution of subprocess: {}" \
-          .format(traceback.format_exc(), args[0][6], error.output)) from error
+          .format(traceback.format_exc(),
+                  _pip_package_from_args(args), error.output)) from error
       else:
         raise RuntimeError("Full trace: {} \
           \n Output of the failed child process: {}" \
@@ -90,11 +105,12 @@ def check_output(*args, **kwargs):
     except OSError as e:
       raise RuntimeError("Executable {} not found".format(args[0])) from e
     except subprocess.CalledProcessError as error:
-      if isinstance(args, tuple) and (args[0][2] == "pip"):
+      if isinstance(args, tuple) and len(args[0]) > 2 and args[0][2] == "pip":
         raise RuntimeError( \
           "Full traceback: {} \n Pip install failed for package: {} \
           \n Output from execution of subprocess: {}" \
-          .format(traceback.format_exc(), args[0][6], error.output)) from error
+          .format(traceback.format_exc(),
+                  _pip_package_from_args(args), error.output)) from error
       else:
         raise RuntimeError("Full trace: {}, \
            output of the failed child process {} "\

sdks/python/apache_beam/utils/processes_test.py

@@ -131,6 +131,19 @@ def test_check_call_pip_install_non_existing_package(self):
       self.assertIn("Pip install failed for package: {}".format(package),\
         error.args[0])
 
+  def test_check_call_pip_short_command_no_index_error(self):
+    """Short pip command (e.g. pip install pkg) must not raise IndexError."""
+    returncode = 1
+    cmd = ['python', '-m', 'pip', 'install', 'nonexistent-package-xyz']
+    output = "ERROR: Could not find a version that satisfies"
+    self.mock_get.side_effect = subprocess.CalledProcessError(
+        returncode, cmd, output=output)
+    with self.assertRaises(RuntimeError) as ctx:
+      processes.check_call(cmd)
+    self.assertIn("Output from execution of subprocess:", ctx.exception.args[0])
+    self.assertIn(output, ctx.exception.args[0])
+    self.assertIn("see output below", ctx.exception.args[0])
+
 
 class TestErrorHandlingCheckOutput(unittest.TestCase):
   @classmethod
@@ -172,6 +185,19 @@ def test_check_output_pip_install_non_existing_package(self):
       self.assertIn("Pip install failed for package: {}".format(package),\
         error.args[0])
 
+  def test_check_output_pip_short_command_no_index_error(self):
+    """Short pip command must not raise IndexError."""
+    returncode = 1
+    cmd = ['python', '-m', 'pip', 'install', 'nonexistent-package-xyz']
+    output = "ERROR: Could not find a version"
+    self.mock_get.side_effect = subprocess.CalledProcessError(
+        returncode, cmd, output=output)
+    with self.assertRaises(RuntimeError) as ctx:
+      processes.check_output(cmd)
+    self.assertIn("Output from execution of subprocess:", ctx.exception.args[0])
+    self.assertIn(output, ctx.exception.args[0])
+    self.assertIn("see output below", ctx.exception.args[0])
+
 
 class TestErrorHandlingCall(unittest.TestCase):
   @classmethod
@@ -213,6 +239,19 @@ def test_check_output_pip_install_non_existing_package(self):
       self.assertIn("Pip install failed for package: {}".format(package),\
         error.args[0])
 
+  def test_call_pip_short_command_no_index_error(self):
+    """Short pip command must not raise IndexError."""
+    returncode = 1
+    cmd = ['python', '-m', 'pip', 'install', 'nonexistent-package-xyz']
+    output = "ERROR: Could not find a version"
+    self.mock_get.side_effect = subprocess.CalledProcessError(
+        returncode, cmd, output=output)
+    with self.assertRaises(RuntimeError) as ctx:
+      processes.call(cmd)
+    self.assertIn("Output from execution of subprocess:", ctx.exception.args[0])
+    self.assertIn(output, ctx.exception.args[0])
+    self.assertIn("see output below", ctx.exception.args[0])
+
 
 if __name__ == '__main__':
   unittest.main()