Fix is_interactive() method

Author: virrrat

google/cloud/dataproc_spark_connect/environment.py

@@ -48,7 +48,15 @@ def is_jetbrains_ide() -> bool:
 
 
 def is_interactive():
-    return hasattr(sys, "ps1")
+    try:
+        from IPython import get_ipython
+
+        if get_ipython() is not None:
+            return True
+    except ImportError:
+        pass
+
+    return hasattr(sys, "ps1") or sys.flags.interactive
 
 
 def is_terminal():

tests/unit/test_environment.py

@@ -222,17 +222,42 @@ def test_get_client_environment_label_precedence(
             "colab-enterprise",
         )
 
+    @mock.patch("IPython.get_ipython", return_value=mock.MagicMock())
+    def test_is_interactive_ipython_true(self, mock_get_ipython):
+        self.assertTrue(environment.is_interactive())
+
+    @mock.patch("IPython.get_ipython", return_value=None)
+    @mock.patch("google.cloud.dataproc_spark_connect.environment.sys")
+    def test_is_interactive_ipython_false(self, mock_sys, mock_get_ipython):
+        if hasattr(mock_sys, "ps1"):
+            del mock_sys.ps1
+        mock_sys.flags.interactive = 0
+        self.assertFalse(environment.is_interactive())
+
+    @mock.patch("IPython.get_ipython", side_effect=ImportError)
     @mock.patch("google.cloud.dataproc_spark_connect.environment.sys")
-    def test_is_interactive_true(self, mock_sys):
+    def test_is_interactive_true_via_ps1(self, mock_sys, mock_get_ipython):
         # Simulate interactive environment by setting ps1
         mock_sys.ps1 = ">>>"
+        mock_sys.flags.interactive = 0
         self.assertTrue(environment.is_interactive())
 
+    @mock.patch("IPython.get_ipython", side_effect=ImportError)
     @mock.patch("google.cloud.dataproc_spark_connect.environment.sys")
-    def test_is_interactive_false(self, mock_sys):
-        # Simulate non-interactive environment by removing ps1
+    def test_is_interactive_true_via_flags(self, mock_sys, mock_get_ipython):
+        # Simulate interactive environment via sys.flags.interactive
         if hasattr(mock_sys, "ps1"):
             del mock_sys.ps1
+        mock_sys.flags.interactive = 1
+        self.assertTrue(environment.is_interactive())
+
+    @mock.patch("IPython.get_ipython", side_effect=ImportError)
+    @mock.patch("google.cloud.dataproc_spark_connect.environment.sys")
+    def test_is_interactive_false(self, mock_sys, mock_get_ipython):
+        # Simulate non-interactive environment
+        if hasattr(mock_sys, "ps1"):
+            del mock_sys.ps1
+        mock_sys.flags.interactive = 0
         self.assertFalse(environment.is_interactive())
 
     @mock.patch("sys.stdin")
@@ -254,9 +279,13 @@ def test_is_interactive_terminal_true(self, mock_sys, mock_stdin):
 
     @mock.patch("sys.stdin")
     @mock.patch("google.cloud.dataproc_spark_connect.environment.sys")
-    def test_is_interactive_terminal_true(self, mock_sys, mock_stdin):
+    @mock.patch("IPython.get_ipython", side_effect=ImportError)
+    def test_is_interactive_terminal_false(
+        self, mock_get_ipython, mock_sys, mock_stdin
+    ):
         if hasattr(mock_sys, "ps1"):
             del mock_sys.ps1
+        mock_sys.flags.interactive = 0
         mock_stdin.isatty.return_value = False
         self.assertFalse(environment.is_interactive_terminal())