feat: allow configurable child shutdown timeouts

nox/popen.py

@@ -21,16 +21,19 @@
 from typing import IO, Mapping, Sequence
 
 
-def shutdown_process(proc: subprocess.Popen) -> tuple[bytes, bytes]:
+def shutdown_process(
+    proc: subprocess.Popen,
+    interrupt_timeout: float,
+    terminate_timeout: float,
+) -> tuple[bytes, bytes]:
     """Gracefully shutdown a child process."""
-
     with contextlib.suppress(subprocess.TimeoutExpired):
-        return proc.communicate(timeout=0.3)
+        return proc.communicate(timeout=interrupt_timeout)
 
     proc.terminate()
 
     with contextlib.suppress(subprocess.TimeoutExpired):
-        return proc.communicate(timeout=0.2)
+        return proc.communicate(timeout=terminate_timeout)
 
     proc.kill()
 
@@ -60,6 +63,8 @@ def popen(
     silent: bool = False,
     stdout: int | IO | None = None,
     stderr: int | IO = subprocess.STDOUT,
+    interrupt_timeout: float = 0.3,
+    terminate_timeout: float = 0.2,
 ) -> tuple[int, str]:
     if silent and stdout is not None:
         raise ValueError(
@@ -77,7 +82,7 @@ def popen(
         sys.stdout.flush()
 
     except KeyboardInterrupt:
-        out, err = shutdown_process(proc)
+        out, err = shutdown_process(proc, interrupt_timeout, terminate_timeout)
         if proc.returncode != 0:
             raise
 

tests/test_sessions.py

@@ -283,7 +283,10 @@ def test_run_external_not_a_virtualenv(self):
             session.run(sys.executable, "--version")
 
         run.assert_called_once_with(
-            (sys.executable, "--version"), external=True, env=mock.ANY, paths=None
+            (sys.executable, "--version"),
+            external=True,
+            env=mock.ANY,
+            paths=None,
         )
 
     def test_run_external_condaenv(self):
@@ -344,6 +347,55 @@ def test_run_always_install_only(self, caplog):
 
         assert session.run_always(operator.add, 23, 19) == 42
 
+    @pytest.mark.parametrize(
+        (
+            "interrupt_timeout_setting",
+            "terminate_timeout_setting",
+            "interrupt_timeout_expected",
+            "terminate_timeout_expected",
+        ),
+        [
+            (None, None, 0.3, 0.2),
+            (0, 0, 0, 0),
+            (1, 2, 1, 2),
+        ],
+    )
+    def test_run_shutdown_process_timeouts(
+        self,
+        interrupt_timeout_setting,
+        terminate_timeout_setting,
+        interrupt_timeout_expected,
+        terminate_timeout_expected,
+    ):
+        session, runner = self.make_session_and_runner()
+
+        runner.venv = nox.virtualenv.ProcessEnv()
+
+        subp_popen_instance = mock.Mock()
+        subp_popen_instance.communicate.side_effect = KeyboardInterrupt()
+        with mock.patch(
+            "nox.popen.shutdown_process", autospec=True
+        ) as shutdown_process, mock.patch(
+            "nox.popen.subprocess.Popen",
+            new=mock.Mock(return_value=subp_popen_instance),
+        ):
+            shutdown_process.return_value = ("", "")
+
+            timeout_kwargs = {}
+            if interrupt_timeout_setting is not None:
+                timeout_kwargs["interrupt_timeout"] = interrupt_timeout_setting
+            if terminate_timeout_setting is not None:
+                timeout_kwargs["terminate_timeout"] = terminate_timeout_setting
+
+            with pytest.raises(KeyboardInterrupt):
+                session.run(sys.executable, "--version", **timeout_kwargs)
+
+        shutdown_process.assert_called_once_with(
+            proc=mock.ANY,
+            interrupt_timeout=interrupt_timeout_expected,
+            terminate_timeout=terminate_timeout_expected,
+        )
+
     @pytest.mark.parametrize(
         ("no_install", "reused", "run_called"),
         [