enable streaming of integration test output (#21912)

Add support to the Pants integration test framework to stream output to
the console. This is very useful when debugging long-running integration
tests which would otherwise show no output while they run since the
integration test framework currently only captures output to a buffer.

Adds a `stream_output` parameter to `PantsJoinHandle.join`,
`run_pants_with_workdir`, and `run_pants`. Enable `stream_output=True`
on a specific invocation of `run_pants_with_workdir` and `run_pants`,
and then run the test with `--debug` so the test is invoked as an
interactive process.

---------

Co-authored-by: Huon Wilson <[email protected]>

docs/docs/writing-plugins/the-rules-api/testing-plugins.mdx

@@ -598,3 +598,9 @@ def test_junit_report() -> None:
     coverage_report = Path(get_buildroot(), "dist", "coverage", "python", "report.json")
     assert coverage_report.read_text() == "foo"
 ```
+
+### Debugging integration tests
+
+While developing and debugging integration tests, you can have Pants stream the output for the Pants invocation under test to the console. This is useful, for example, when debugging long-running integration tests which would otherwise show no output while they run.
+
+To use, adjust specific test(s) to use the `stream_output` parameter, for example, `run_pants_with_workdir(..., stream_output=True)` or `run_pants(..., stream_output=True)`, and then run the test with `pants test --debug path/to:test -- --capture=no` so the test is invoked as an interactive process and pytest does not capture the output during the run.

docs/notes/2.25.x.md

@@ -209,6 +209,7 @@ The version of Python used by Pants itself is now [3.11](https://docs.python.org
 
 The oldest [glibc version](https://www.sourceware.org/glibc/wiki/Glibc%20Timeline) supported by the published Pants wheels is now 2.28.  This should have no effect unless you are running on extremely old Linux distributions.  See <https://github.com/pypa/manylinux> for background context on Python wheels and C libraries.
 
+The integration testing framework in the `pantsbuild.pants.testutil` package now supports streaming the output of the Pants invocation under test to the console. This is useful when debugging long-running integration tests which would otherwise show no output while they run since the integration test framework previously only captured output to a buffer. To use, adjust specific test(s) to use the new `stream_output` parameter, for example, `run_pants_with_workdir(..., stream_output=True)` or `run_pants(..., stream_output=True)`, and then run the test with `pants test --debug path/to:test -- --capture=no` so the test is invoked as an interactive process and pytest does not capture the output during the run.
 
 ## Full Changelog
 

src/python/pants/testutil/pants_integration_test.py

@@ -3,13 +3,16 @@
 
 from __future__ import annotations
 
+import errno
 import glob
 import os
 import subprocess
 import sys
 from contextlib import contextmanager
 from dataclasses import dataclass
-from typing import Any, Iterator, List, Mapping, Union, cast
+from io import BytesIO
+from threading import Thread
+from typing import Any, Iterator, List, Mapping, TextIO, Union, cast
 
 import pytest
 import toml
@@ -67,13 +70,72 @@ class PantsJoinHandle:
     process: subprocess.Popen
     workdir: str
 
-    def join(self, stdin_data: bytes | str | None = None) -> PantsResult:
+    # Write data to the child's stdin pipe and then close the pipe. (Copied from Python source
+    # at https://github.com/python/cpython/blob/e41ec8e18b078024b02a742272e675ae39778536/Lib/subprocess.py#L1151
+    # to handle the same edge cases handled by `subprocess.Popen.communicate`.)
+    def _stdin_write(self, input: bytes | str | None):
+        assert self.process.stdin
+
+        if input:
+            try:
+                binary_input = ensure_binary(input)
+                self.process.stdin.write(binary_input)
+            except BrokenPipeError:
+                pass  # communicate() must ignore broken pipe errors.
+            except OSError as exc:
+                if exc.errno == errno.EINVAL:
+                    # bpo-19612, bpo-30418: On Windows, stdin.write() fails
+                    # with EINVAL if the child process exited or if the child
+                    # process is still running but closed the pipe.
+                    pass
+                else:
+                    raise
+
+        try:
+            self.process.stdin.close()
+        except BrokenPipeError:
+            pass  # communicate() must ignore broken pipe errors.
+        except OSError as exc:
+            if exc.errno == errno.EINVAL:
+                pass
+            else:
+                raise
+
+    def join(
+        self, stdin_data: bytes | str | None = None, stream_output: bool = False
+    ) -> PantsResult:
         """Wait for the pants process to complete, and return a PantsResult for it."""
-        if stdin_data is not None:
-            stdin_data = ensure_binary(stdin_data)
-        (stdout, stderr) = self.process.communicate(stdin_data)
 
-        if self.process.returncode != PANTS_SUCCEEDED_EXIT_CODE:
+        def worker(in_stream: BytesIO, buffer: bytearray, out_stream: TextIO) -> None:
+            while data := in_stream.read1(1024):
+                buffer.extend(data)
+                out_stream.write(data.decode(errors="ignore"))
+                out_stream.flush()
+
+        if stream_output:
+            stdout_buffer = bytearray()
+            stdout_thread = Thread(
+                target=worker, args=(self.process.stdout, stdout_buffer, sys.stdout)
+            )
+            stdout_thread.daemon = True
+            stdout_thread.start()
+
+            stderr_buffer = bytearray()
+            stderr_thread = Thread(
+                target=worker, args=(self.process.stderr, stderr_buffer, sys.stderr)
+            )
+            stderr_thread.daemon = True
+            stderr_thread.start()
+
+            self._stdin_write(stdin_data)
+            self.process.wait()
+            stdout, stderr = (bytes(stdout_buffer), bytes(stderr_buffer))
+        else:
+            if stdin_data is not None:
+                stdin_data = ensure_binary(stdin_data)
+            stdout, stderr = self.process.communicate(stdin_data)
+
+        if self.process.returncode != PANTS_SUCCEEDED_EXIT_CODE or stream_output:
             render_logs(self.workdir)
 
         return PantsResult(
@@ -202,6 +264,7 @@ def run_pants_with_workdir(
     stdin_data: bytes | str | None = None,
     shell: bool = False,
     set_pants_ignore: bool = True,
+    stream_output: bool = False,
 ) -> PantsResult:
     handle = run_pants_with_workdir_without_waiting(
         command,
@@ -213,7 +276,7 @@ def run_pants_with_workdir(
         extra_env=extra_env,
         set_pants_ignore=set_pants_ignore,
     )
-    return handle.join(stdin_data=stdin_data)
+    return handle.join(stdin_data=stdin_data, stream_output=stream_output)
 
 
 def run_pants(
@@ -224,6 +287,7 @@ def run_pants(
     config: Mapping | None = None,
     extra_env: Env | None = None,
     stdin_data: bytes | str | None = None,
+    stream_output: bool = False,
 ) -> PantsResult:
     """Runs Pants in a subprocess.
 
@@ -244,6 +308,7 @@ def run_pants(
             config=config,
             stdin_data=stdin_data,
             extra_env=extra_env,
+            stream_output=stream_output,
         )