diff --git a/b2/_internal/_cli/autocomplete_install.py b/b2/_internal/_cli/autocomplete_install.py index 12a6b99e..57b08f50 100644 --- a/b2/_internal/_cli/autocomplete_install.py +++ b/b2/_internal/_cli/autocomplete_install.py @@ -7,8 +7,12 @@ # License https://www.backblaze.com/using_b2_code.html # ###################################################################### +from __future__ import annotations + import abc import logging +import os +import pty import re import shutil import subprocess @@ -16,7 +20,6 @@ from datetime import datetime from pathlib import Path from shlex import quote -from typing import List import argcomplete from class_registry import ClassRegistry, RegistryKeyError @@ -91,7 +94,7 @@ def get_script_path(self) -> Path: def program_in_path(self) -> bool: """Check if the given program is in PATH.""" - return _silent_success_run([self.shell_exec, '-c', self.prog]) + return _silent_pty_run([self.shell_exec, '-c', self.prog]) @abc.abstractmethod def is_enabled(self) -> bool: @@ -130,7 +133,7 @@ def get_script_path(self) -> Path: def is_enabled(self) -> bool: """Check if autocompletion is enabled.""" - return _silent_success_run([self.shell_exec, '-i', '-c', f'complete -p {quote(self.prog)}']) + return _silent_pty_run([self.shell_exec, '-i', '-c', f'complete -p {quote(self.prog)}']) @SHELL_REGISTRY.register('bash') @@ -157,7 +160,7 @@ def get_script_path(self) -> Path: return Path("~/.zsh/completion/").expanduser() / f"_{self.prog}" def is_enabled(self) -> bool: - return _silent_success_run( + return _silent_pty_run( [self.shell_exec, '-i', '-c', f'[[ -v _comps[{quote(self.prog)}] ]]'] ) @@ -185,14 +188,17 @@ def is_enabled(self) -> bool: return self.get_script_path().exists() -def _silent_success_run(cmd: List[str]) -> bool: - p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) +def _silent_pty_run(cmd: list[str]) -> bool: + master_fd, stdin = pty.openpty() + p = subprocess.Popen(cmd, stdin=stdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + os.close(stdin) stdout, stderr = p.communicate() if p.returncode != 0: logger.debug( "Command %s exited with code %d, stdout: %s, stderr: %s", cmd, p.returncode, stdout, stderr ) + os.close(master_fd) return p.returncode == 0