[stable-2.13] ansible-test - Fix TTY and output handling. (#78350).

(cherry picked from commit a3c90dd0bcb4aecfc64a4a584e52aec77ee61158)

Co-authored-by: Matt Clay <matt@mystile.com>

8 files changed, 64 insertions, 34 deletions

diff --git a/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/__init__.py b/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/__init__.py
index 412414050f..267969886e 100644
--- a/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/__init__.py
+++ b/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/__init__.py
@@ -29,10 +29,6 @@ TargetSetIndexes = t.Dict[t.FrozenSet[int], int]
 
 class CoverageAnalyzeTargetsConfig(CoverageAnalyzeConfig):
     """Configuration for the `coverage analyze targets` command."""
-    def __init__(self, args):  # type: (t.Any) -> None
-        super().__init__(args)
-
-        self.display_stderr = True
 
 
 def make_report(target_indexes, arcs, lines):  # type: (TargetIndexes, Arcs, Lines) -> t.Dict[str, t.Any]
diff --git a/test/lib/ansible_test/_internal/commands/shell/__init__.py b/test/lib/ansible_test/_internal/commands/shell/__init__.py
index 4bd3fde02a..e62437ead7 100644
--- a/test/lib/ansible_test/_internal/commands/shell/__init__.py
+++ b/test/lib/ansible_test/_internal/commands/shell/__init__.py
@@ -7,6 +7,7 @@ import typing as t
 
 from ...util import (
     ApplicationError,
+    OutputStream,
     display,
 )
 
@@ -82,7 +83,10 @@ def command_shell(args):  # type: (ShellConfig) -> None
         return
 
     if args.cmd:
-        con.run(args.cmd, capture=False, interactive=False, force_stdout=True)
+        # Running a command is assumed to be non-interactive. Only a shell (no command) is interactive.
+        # If we want to support interactive commands in the future, we'll need an `--interactive` command line option.
+        # Command stderr output is allowed to mix with our own output, which is all sent to stderr.
+        con.run(args.cmd, capture=False, interactive=False, output_stream=OutputStream.ORIGINAL)
         return
 
     if isinstance(con, SshConnection) and args.raw:
diff --git a/test/lib/ansible_test/_internal/config.py b/test/lib/ansible_test/_internal/config.py
index 1e08cfb478..27395678c7 100644
--- a/test/lib/ansible_test/_internal/config.py
+++ b/test/lib/ansible_test/_internal/config.py
@@ -217,7 +217,7 @@ class ShellConfig(EnvironmentConfig):
         self.cmd = args.cmd  # type: t.List[str]
         self.raw = args.raw  # type: bool
         self.check_layout = self.delegate  # allow shell to be used without a valid layout as long as no delegation is required
-        self.interactive = True
+        self.interactive = sys.stdin.isatty() and not args.cmd  # delegation should only be interactive when stdin is a TTY and no command was given
         self.export = args.export  # type: t.Optional[str]
         self.display_stderr = True
 
diff --git a/test/lib/ansible_test/_internal/connections.py b/test/lib/ansible_test/_internal/connections.py
index 026058abb4..95ad7331a7 100644
--- a/test/lib/ansible_test/_internal/connections.py
+++ b/test/lib/ansible_test/_internal/connections.py
@@ -16,6 +16,7 @@ from .config import (
 
 from .util import (
     Display,
+    OutputStream,
     SubprocessError,
     retry,
 )
@@ -50,7 +51,7 @@ class Connection(metaclass=abc.ABCMeta):
             data=None,  # type: t.Optional[str]
             stdin=None,  # type: t.Optional[t.IO[bytes]]
             stdout=None,  # type: t.Optional[t.IO[bytes]]
-            force_stdout=False,  # type: bool
+            output_stream=None,  # type: t.Optional[OutputStream]
             ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
         """Run the specified command and return the result."""
 
@@ -98,7 +99,7 @@ class LocalConnection(Connection):
             data=None,  # type: t.Optional[str]
             stdin=None,  # type: t.Optional[t.IO[bytes]]
             stdout=None,  # type: t.Optional[t.IO[bytes]]
-            force_stdout=False,  # type: bool
+            output_stream=None,  # type: t.Optional[OutputStream]
             ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
         """Run the specified command and return the result."""
         return run_command(
@@ -109,7 +110,7 @@ class LocalConnection(Connection):
             stdin=stdin,
             stdout=stdout,
             interactive=interactive,
-            force_stdout=force_stdout,
+            output_stream=output_stream,
         )
 
 
@@ -140,7 +141,7 @@ class SshConnection(Connection):
             data=None,  # type: t.Optional[str]
             stdin=None,  # type: t.Optional[t.IO[bytes]]
             stdout=None,  # type: t.Optional[t.IO[bytes]]
-            force_stdout=False,  # type: bool
+            output_stream=None,  # type: t.Optional[OutputStream]
             ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
         """Run the specified command and return the result."""
         options = list(self.options)
@@ -174,7 +175,7 @@ class SshConnection(Connection):
                 stdin=stdin,
                 stdout=stdout,
                 interactive=interactive,
-                force_stdout=force_stdout,
+                output_stream=output_stream,
                 error_callback=error_callback,
             )
 
@@ -222,7 +223,7 @@ class DockerConnection(Connection):
             data=None,  # type: t.Optional[str]
             stdin=None,  # type: t.Optional[t.IO[bytes]]
             stdout=None,  # type: t.Optional[t.IO[bytes]]
-            force_stdout=False,  # type: bool
+            output_stream=None,  # type: t.Optional[OutputStream]
             ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
         """Run the specified command and return the result."""
         options = []
@@ -243,7 +244,7 @@ class DockerConnection(Connection):
             stdin=stdin,
             stdout=stdout,
             interactive=interactive,
-            force_stdout=force_stdout,
+            output_stream=output_stream,
         )
 
     def inspect(self):  # type: () -> DockerInspect
diff --git a/test/lib/ansible_test/_internal/delegation.py b/test/lib/ansible_test/_internal/delegation.py
index 79d2dc7cbc..975b6fc7e5 100644
--- a/test/lib/ansible_test/_internal/delegation.py
+++ b/test/lib/ansible_test/_internal/delegation.py
@@ -27,6 +27,7 @@ from .util import (
     ANSIBLE_BIN_PATH,
     ANSIBLE_LIB_ROOT,
     ANSIBLE_TEST_ROOT,
+    OutputStream,
 )
 
 from .util_common import (
@@ -191,7 +192,12 @@ def delegate_command(args, host_state, exclude, require):  # type: (EnvironmentC
         success = False
 
         try:
-            con.run(insert_options(command, options), capture=False, interactive=args.interactive)
+            # When delegating, preserve the original separate stdout/stderr streams, but only when the following conditions are met:
+            # 1) Display output is being sent to stderr. This indicates the output on stdout must be kept separate from stderr.
+            # 2) The delegation is non-interactive. Interactive mode, which generally uses a TTY, is not compatible with intercepting stdout/stderr.
+            # The downside to having separate streams is that individual lines of output from each are more likely to appear out-of-order.
+            output_stream = OutputStream.ORIGINAL if args.display_stderr and not args.interactive else None
+            con.run(insert_options(command, options), capture=False, interactive=args.interactive, output_stream=output_stream)
             success = True
         finally:
             if host_delegation:
diff --git a/test/lib/ansible_test/_internal/docker_util.py b/test/lib/ansible_test/_internal/docker_util.py
index 0a17bf4c66..2ed6504bcb 100644
--- a/test/lib/ansible_test/_internal/docker_util.py
+++ b/test/lib/ansible_test/_internal/docker_util.py
@@ -20,6 +20,7 @@ from .util import (
     find_executable,
     SubprocessError,
     cache,
+    OutputStream,
 )
 
 from .util_common import (
@@ -515,7 +516,7 @@ def docker_exec(
         stdin=None,  # type: t.Optional[t.IO[bytes]]
         stdout=None,  # type: t.Optional[t.IO[bytes]]
         interactive=False,  # type: bool
-        force_stdout=False,  # type: bool
+        output_stream=None,  # type: t.Optional[OutputStream]
         data=None,  # type: t.Optional[str]
 ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
     """Execute the given command in the specified container."""
@@ -526,7 +527,7 @@ def docker_exec(
         options.append('-i')
 
     return docker_command(args, ['exec'] + options + [container_id] + cmd, capture=capture, stdin=stdin, stdout=stdout, interactive=interactive,
-                          force_stdout=force_stdout, data=data)
+                          output_stream=output_stream, data=data)
 
 
 def docker_info(args):  # type: (CommonConfig) -> t.Dict[str, t.Any]
@@ -548,7 +549,7 @@ def docker_command(
         stdin=None,  # type: t.Optional[t.IO[bytes]]
         stdout=None,  # type: t.Optional[t.IO[bytes]]
         interactive=False,  # type: bool
-        force_stdout=False,  # type: bool
+        output_stream=None,  # type: t.Optional[OutputStream]
         always=False,  # type: bool
         data=None,  # type: t.Optional[str]
 ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
@@ -560,7 +561,7 @@ def docker_command(
         command.append('--remote')
 
     return run_command(args, command + cmd, env=env, capture=capture, stdin=stdin, stdout=stdout, interactive=interactive, always=always,
-                       force_stdout=force_stdout, data=data)
+                       output_stream=output_stream, data=data)
 
 
 def docker_environment():  # type: () -> t.Dict[str, str]
diff --git a/test/lib/ansible_test/_internal/util.py b/test/lib/ansible_test/_internal/util.py
index 25cca59d40..60901ef507 100644
--- a/test/lib/ansible_test/_internal/util.py
+++ b/test/lib/ansible_test/_internal/util.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 
 import abc
 import errno
+import enum
 import fcntl
 import importlib.util
 import inspect
@@ -100,6 +101,24 @@ MODE_DIRECTORY = MODE_READ | stat.S_IWUSR | stat.S_IXUSR | stat.S_IXGRP | stat.S
 MODE_DIRECTORY_WRITE = MODE_DIRECTORY | stat.S_IWGRP | stat.S_IWOTH
 
 
+class OutputStream(enum.Enum):
+    """The output stream to use when running a subprocess and redirecting/capturing stdout or stderr."""
+
+    ORIGINAL = enum.auto()
+    AUTO = enum.auto()
+
+    def get_buffer(self, original: t.BinaryIO) -> t.BinaryIO:
+        """Return the correct output buffer to use, taking into account the given original buffer."""
+
+        if self == OutputStream.ORIGINAL:
+            return original
+
+        if self == OutputStream.AUTO:
+            return display.fd.buffer
+
+        raise NotImplementedError(str(self))
+
+
 class Architecture:
     """
     Normalized architecture names.
@@ -330,12 +349,14 @@ def raw_command(
         stdin=None,  # type: t.Optional[t.Union[t.IO[bytes], int]]
         stdout=None,  # type: t.Optional[t.Union[t.IO[bytes], int]]
         interactive=False,  # type: bool
-        force_stdout=False,  # type: bool
+        output_stream=None,  # type: t.Optional[OutputStream]
         cmd_verbosity=1,  # type: int
         str_errors='strict',  # type: str
         error_callback=None,  # type: t.Optional[t.Callable[[SubprocessError], None]]
 ):  # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
     """Run the specified command and return stdout and stderr as a tuple."""
+    output_stream = output_stream or OutputStream.AUTO
+
     if capture and interactive:
         raise InternalError('Cannot combine capture=True with interactive=True.')
 
@@ -354,11 +375,11 @@ def raw_command(
     if stdout and not capture:
         raise InternalError('Redirection of stdout requires capture=True to avoid redirection of stderr to stdout.')
 
-    if force_stdout and capture:
-        raise InternalError('Cannot combine force_stdout=True with capture=True.')
+    if output_stream != OutputStream.AUTO and capture:
+        raise InternalError(f'Cannot combine {output_stream=} with capture=True.')
 
-    if force_stdout and interactive:
-        raise InternalError('Cannot combine force_stdout=True with interactive=True.')
+    if output_stream != OutputStream.AUTO and interactive:
+        raise InternalError(f'Cannot combine {output_stream=} with interactive=True.')
 
     if not cwd:
         cwd = os.getcwd()
@@ -425,9 +446,9 @@ def raw_command(
         # This prevents subprocesses from sharing stdout/stderr with the current process or each other.
         # Doing so allows subprocesses to safely make changes to their file handles, such as making them non-blocking (ssh does this).
         # This also maintains consistency between local testing and CI systems, which typically do not provide a TTY.
-        # To maintain output ordering, a single pipe is used for both stdout/stderr when not capturing output.
+        # To maintain output ordering, a single pipe is used for both stdout/stderr when not capturing output unless the output stream is ORIGINAL.
         stdout = stdout or subprocess.PIPE
-        stderr = subprocess.PIPE if capture else subprocess.STDOUT
+        stderr = subprocess.PIPE if capture or output_stream == OutputStream.ORIGINAL else subprocess.STDOUT
         communicate = True
     else:
         stderr = None
@@ -448,7 +469,7 @@ def raw_command(
         if communicate:
             data_bytes = to_optional_bytes(data)
             stdout_bytes, stderr_bytes = communicate_with_process(process, data_bytes, stdout == subprocess.PIPE, stderr == subprocess.PIPE, capture=capture,
-                                                                  force_stdout=force_stdout)
+                                                                  output_stream=output_stream)
             stdout_text = to_optional_text(stdout_bytes, str_errors) or u''
             stderr_text = to_optional_text(stderr_bytes, str_errors) or u''
         else:
@@ -477,7 +498,7 @@ def communicate_with_process(
         stdout: bool,
         stderr: bool,
         capture: bool,
-        force_stdout: bool
+        output_stream: OutputStream,
 ) -> t.Tuple[bytes, bytes]:
     """Communicate with the specified process, handling stdin/stdout/stderr as requested."""
     threads: t.List[WrappedThread] = []
@@ -492,13 +513,13 @@ def communicate_with_process(
         threads.append(WriterThread(process.stdin, stdin))
 
     if stdout:
-        stdout_reader = reader(process.stdout, force_stdout)
+        stdout_reader = reader(process.stdout, output_stream.get_buffer(sys.stdout.buffer))
         threads.append(stdout_reader)
     else:
         stdout_reader = None
 
     if stderr:
-        stderr_reader = reader(process.stderr, force_stdout)
+        stderr_reader = reader(process.stderr, output_stream.get_buffer(sys.stderr.buffer))
         threads.append(stderr_reader)
     else:
         stderr_reader = None
@@ -546,11 +567,11 @@ class WriterThread(WrappedThread):
 
 class ReaderThread(WrappedThread, metaclass=abc.ABCMeta):
     """Thread to read stdout from a subprocess."""
-    def __init__(self, handle: t.IO[bytes], force_stdout: bool) -> None:
+    def __init__(self, handle: t.IO[bytes], buffer: t.BinaryIO) -> None:
         super().__init__(self._run)
 
         self.handle = handle
-        self.force_stdout = force_stdout
+        self.buffer = buffer
         self.lines = []  # type: t.List[bytes]
 
     @abc.abstractmethod
@@ -577,7 +598,7 @@ class OutputThread(ReaderThread):
     def _run(self) -> None:
         """Workload to run on a thread."""
         src = self.handle
-        dst = sys.stdout.buffer if self.force_stdout else display.fd.buffer
+        dst = self.buffer
 
         try:
             for line in src:
diff --git a/test/lib/ansible_test/_internal/util_common.py b/test/lib/ansible_test/_internal/util_common.py
index 86e0d776bb..b49f5d48f0 100644
--- a/test/lib/ansible_test/_internal/util_common.py
+++ b/test/lib/ansible_test/_internal/util_common.py
@@ -27,6 +27,7 @@ from .util import (
     MODE_DIRECTORY,
     MODE_FILE_EXECUTE,
     MODE_FILE,
+    OutputStream,
     PYTHON_PATHS,
     raw_command,
     ANSIBLE_TEST_DATA_ROOT,
@@ -409,7 +410,7 @@ def run_command(
         stdin=None,  # type: t.Optional[t.IO[bytes]]
         stdout=None,  # type: t.Optional[t.IO[bytes]]
         interactive=False,  # type: bool
-        force_stdout=False,  # type: bool
+        output_stream=None,  # type: t.Optional[OutputStream]
         cmd_verbosity=1,  # type: int
         str_errors='strict',  # type: str
         error_callback=None,  # type: t.Optional[t.Callable[[SubprocessError], None]]
@@ -417,7 +418,7 @@ def run_command(
     """Run the specified command and return stdout and stderr as a tuple."""
     explain = args.explain and not always
     return raw_command(cmd, capture=capture, env=env, data=data, cwd=cwd, explain=explain, stdin=stdin, stdout=stdout, interactive=interactive,
-                       force_stdout=force_stdout, cmd_verbosity=cmd_verbosity, str_errors=str_errors, error_callback=error_callback)
+                       output_stream=output_stream, cmd_verbosity=cmd_verbosity, str_errors=str_errors, error_callback=error_callback)
 
 
 def yamlcheck(python):