Fix container --tty detection in subprocess mode (#1306)

Check input and output file descriptors for tty for run_command() API in subprocess path. The pexpect path is left unchanged.

---
Signed-off-by: xz-dev <xiangzhedev@gmail.com>

Author: xz-dev

src/ansible_runner/config/_base.py

@@ -487,6 +487,15 @@ def _handle_ansible_cmd_options_bind_mounts(self, args_list: list[str], cmdline_
 
             self._update_volume_mount_paths(args_list, optional_arg_value)
 
+    def should_allocate_tty(self) -> bool:
+        """Whether the container should get a ``--tty`` flag.
+
+        The base implementation returns ``False``.
+        Subclasses (e.g. :class:`~ansible_runner.config.command.CommandConfig`)
+        may override this to implement richer logic.
+        """
+        return False
+
     def wrap_args_for_containerization(self,
                                        args: list[str],
                                        execution_mode: BaseExecutionMode,
@@ -495,7 +504,7 @@ def wrap_args_for_containerization(self,
         new_args = [self.process_isolation_executable]
         new_args.extend(['run', '--rm'])
 
-        if self.runner_mode == 'pexpect' or getattr(self, 'input_fd', False):
+        if self.runner_mode == 'pexpect' or self.should_allocate_tty():
             new_args.extend(['--tty'])
 
         new_args.append('--interactive')

src/ansible_runner/config/command.py

@@ -69,6 +69,20 @@ def __init__(self, input_fd=None, output_fd=None, error_fd=None, runner_mode=Non
         'ansible-galaxy',
     )
 
+    def should_allocate_tty(self) -> bool:
+        """Allocate a TTY only when *both* input and output fds are real terminals.
+
+        If either side is redirected (file, pipe, ``/dev/null``) the container
+        must not get ``--tty``, otherwise tools like ``less`` hang or ANSI
+        escapes pollute the redirected output.
+
+        See: https://github.com/ansible/ansible-navigator/issues/1607
+        """
+        return (
+            self.input_fd is not None and hasattr(self.input_fd, 'isatty') and self.input_fd.isatty()
+            and self.output_fd is not None and hasattr(self.output_fd, 'isatty') and self.output_fd.isatty()
+        )
+
     def _set_runner_mode(self):
         if self.input_fd is not None or self.executable_cmd.split(os.pathsep)[-1] in CommandConfig._ANSIBLE_NON_INERACTIVE_CMDS:
             self.runner_mode = 'subprocess'

test/integration/containerized/test_container_management.py

@@ -1,4 +1,5 @@
 import os
+import pty
 import time
 import json
 
@@ -7,7 +8,7 @@
 
 import pytest
 
-from ansible_runner.interface import run
+from ansible_runner.interface import run, run_command
 
 
 @pytest.mark.test_all_runtimes
@@ -182,3 +183,72 @@ def test_registry_auth_file_cleanup(tmp_path, cli, runtime):
     discovered_registry_files = set(glob(auth_registry_glob)) - registry_files_before
     for file_name in discovered_registry_files:
         assert this_ident not in file_name
+
+
+@pytest.mark.test_all_runtimes
+@pytest.mark.parametrize(
+    ('stdin_is_tty', 'stdout_is_tty'),
+    (
+        pytest.param(False, False, id='piped-stdin'),
+        pytest.param(True, False, id='stdout-redirected'),
+        pytest.param(None, None, id='no-fd-headless'),
+    ),
+)
+def test_containerized_tty_allocation(
+    tmp_path, runtime, container_image, stdin_is_tty, stdout_is_tty,
+):
+    """Regression for ansible-navigator#1607: --tty must not appear when fds are not real TTYs.
+
+    The positive case (both fds are TTYs -> --tty is added) is already
+    covered by the unit test
+    ``test_prepare_run_command_containerized_tty_allocation[interactive-tty-*]``
+    which asserts ``'--tty' in rc.command``.  Testing it at the integration
+    level would require a pager (``less``) inside the container image and
+    complex pty plumbing, so we only verify the "no --tty" cases here.
+    """
+    fds_to_close = []
+    kwargs = {
+        'executable_cmd': 'ansible-config',
+        'cmdline_args': ['init'],
+        'private_data_dir': str(tmp_path),
+        'process_isolation': True,
+        'process_isolation_executable': runtime,
+        'container_image': container_image,
+    }
+
+    if stdin_is_tty is not None:
+        if stdin_is_tty:
+            master, slave = pty.openpty()
+            kwargs['input_fd'] = os.fdopen(slave, 'r')
+            fds_to_close.append(('fd', kwargs['input_fd']))
+            fds_to_close.append(('raw', master))
+        else:
+            input_path = tmp_path / 'stdin.txt'
+            input_path.write_text('')
+            kwargs['input_fd'] = input_path.open('r', encoding='utf-8')
+            fds_to_close.append(('fd', kwargs['input_fd']))
+
+    if stdout_is_tty is not None:
+        kwargs['output_fd'] = (tmp_path / 'stdout.txt').open('w', encoding='utf-8')
+        fds_to_close.append(('fd', kwargs['output_fd']))
+
+    if stdin_is_tty is not None or stdout_is_tty is not None:
+        kwargs['error_fd'] = (tmp_path / 'stderr.txt').open('w', encoding='utf-8')
+        fds_to_close.append(('fd', kwargs['error_fd']))
+
+    try:
+        response, _, rc = run_command(**kwargs)
+    finally:
+        for kind, fd in fds_to_close:
+            if kind == 'fd':
+                fd.close()
+            else:
+                os.close(fd)
+
+    if stdin_is_tty is None:
+        content = response
+    else:
+        content = (tmp_path / 'stdout.txt').read_text(encoding='utf-8')
+    assert rc == 0
+    assert '[defaults]' in content
+    assert '\x1b' not in content

test/unit/config/test_command.py

@@ -124,3 +124,45 @@ def test_prepare_run_command_with_containerization(tmp_path, runtime, mocker):
     expected_command_start.extend(cmdline_args)
 
     assert expected_command_start == rc.command
+
+
+@pytest.mark.parametrize('runtime', ('docker', 'podman'))
+@pytest.mark.parametrize(
+    ('stdin_is_tty', 'stdout_is_tty', 'expect_tty'),
+    (
+        pytest.param(False, False, False, id='piped-stdin'),
+        pytest.param(True, True, True, id='interactive-tty'),
+        pytest.param(True, False, False, id='stdout-redirected'),
+        pytest.param(None, None, False, id='no-fd-headless'),
+    ),
+)
+def test_prepare_run_command_containerized_tty_allocation(
+    tmp_path, runtime, mocker, stdin_is_tty, stdout_is_tty, expect_tty,
+):
+    """Regression for ansible-navigator#1607: --tty must only appear when both fds are real TTYs."""
+    mocker.patch.dict('os.environ', {'HOME': str(tmp_path)}, clear=True)
+    tmp_path.joinpath('.ssh').mkdir()
+
+    kwargs = {
+        'private_data_dir': tmp_path,
+        'process_isolation': True,
+        'container_image': 'my_container',
+        'process_isolation_executable': runtime,
+    }
+
+    if stdin_is_tty is not None:
+        mock_stdin = mocker.Mock()
+        mock_stdin.isatty.return_value = stdin_is_tty
+        kwargs['input_fd'] = mock_stdin
+    if stdout_is_tty is not None:
+        mock_stdout = mocker.Mock()
+        mock_stdout.isatty.return_value = stdout_is_tty
+        kwargs['output_fd'] = mock_stdout
+
+    rc = CommandConfig(**kwargs)
+    rc.ident = 'foo'
+    rc.prepare_run_command('ansible-config', cmdline_args=['init'])
+
+    assert rc.runner_mode == 'subprocess'
+    assert ('--tty' in rc.command) == expect_tty
+    assert '--interactive' in rc.command