From 4d409888762ca9ca0ae2d67153be5f21a77f5149 Mon Sep 17 00:00:00 2001 From: Martin Krizek Date: Tue, 5 Sep 2023 21:08:13 +0200 Subject: [PATCH] Add type hints to ansible.utils.display::Display (#81400) * Add type hints to ansible.utils.display::Display Fixes #80841 * Avoid circular import * Fix sanity * type hint some of the functions of the module? * Fix units * Not sure about this * Fix some of the issues from reviews * Add changelog * ... * Update lib/ansible/utils/display.py Co-authored-by: Sviatoslav Sydorenko * remove py2 boilerplate --------- Co-authored-by: Sviatoslav Sydorenko --- .../80841-display-type-annotation.yml | 4 + lib/ansible/executor/playbook_executor.py | 2 +- lib/ansible/playbook/play_context.py | 4 +- lib/ansible/plugins/connection/ssh.py | 2 +- lib/ansible/utils/display.py | 147 +++++++++++------- lib/ansible/utils/encrypt.py | 2 +- lib/ansible/vars/plugins.py | 2 +- 7 files changed, 104 insertions(+), 59 deletions(-) create mode 100644 changelogs/fragments/80841-display-type-annotation.yml diff --git a/changelogs/fragments/80841-display-type-annotation.yml b/changelogs/fragments/80841-display-type-annotation.yml new file mode 100644 index 00000000000000..09eff3ddb3c563 --- /dev/null +++ b/changelogs/fragments/80841-display-type-annotation.yml @@ -0,0 +1,4 @@ +minor_changes: + - Add Python type hints to the Display class (https://github.com/ansible/ansible/issues/80841) +bugfixes: + - vars_prompt - internally convert the ``unsafe`` value to ``bool`` diff --git a/lib/ansible/executor/playbook_executor.py b/lib/ansible/executor/playbook_executor.py index 3feb971d7749f3..52ad0c09a114b0 100644 --- a/lib/ansible/executor/playbook_executor.py +++ b/lib/ansible/executor/playbook_executor.py @@ -148,7 +148,7 @@ def run(self): encrypt = var.get("encrypt", None) salt_size = var.get("salt_size", None) salt = var.get("salt", None) - unsafe = var.get("unsafe", None) + unsafe = boolean(var.get("unsafe", False)) if vname not in self._variable_manager.extra_vars: if self._tqm: diff --git a/lib/ansible/playbook/play_context.py b/lib/ansible/playbook/play_context.py index 1235a81fb92467..d9bb040ed0e810 100644 --- a/lib/ansible/playbook/play_context.py +++ b/lib/ansible/playbook/play_context.py @@ -119,7 +119,7 @@ class PlayContext(Base): def verbosity(self): display.deprecated( "PlayContext.verbosity is deprecated, use ansible.utils.display.Display.verbosity instead.", - version=2.18 + version="2.18" ) return self._internal_verbosity @@ -127,7 +127,7 @@ def verbosity(self): def verbosity(self, value): display.deprecated( "PlayContext.verbosity is deprecated, use ansible.utils.display.Display.verbosity instead.", - version=2.18 + version="2.18" ) self._internal_verbosity = value diff --git a/lib/ansible/plugins/connection/ssh.py b/lib/ansible/plugins/connection/ssh.py index ed43f770bcecc7..49b2ed22fcb858 100644 --- a/lib/ansible/plugins/connection/ssh.py +++ b/lib/ansible/plugins/connection/ssh.py @@ -528,7 +528,7 @@ def wrapped(self: Connection, *args: P.args, **kwargs: P.kwargs) -> tuple[int, b if self._play_context.no_log: display.vvv(u'rc=%s, stdout and stderr censored due to no log' % return_tuple[0], host=self.host) else: - display.vvv(return_tuple, host=self.host) + display.vvv(str(return_tuple), host=self.host) # 0 = success # 1-254 = remote command return code # 255 could be a failure from the ssh command itself diff --git a/lib/ansible/utils/display.py b/lib/ansible/utils/display.py index 0f65eca8d93c05..c7e09dc2b98480 100644 --- a/lib/ansible/utils/display.py +++ b/lib/ansible/utils/display.py @@ -15,8 +15,7 @@ # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type +from __future__ import annotations try: import curses @@ -26,6 +25,7 @@ # this will be set to False if curses.setupterm() fails HAS_CURSES = True +import collections.abc as c import codecs import ctypes.util import fcntl @@ -55,6 +55,10 @@ from ansible.utils.singleton import Singleton from ansible.utils.unsafe_proxy import wrap_var +if t.TYPE_CHECKING: + # avoid circular import at runtime + from ansible.executor.task_queue_manager import FinalQueue + _LIBC = ctypes.cdll.LoadLibrary(ctypes.util.find_library('c')) # Set argtypes, to avoid segfault if the wrong type is provided, # restype is assumed to be c_int @@ -67,7 +71,7 @@ CLEAR_TO_EOL = b'\x1b[K' -def get_text_width(text): +def get_text_width(text: str) -> int: """Function that utilizes ``wcswidth`` or ``wcwidth`` to determine the number of columns used to display a text string. @@ -192,7 +196,7 @@ def filter(self, record): ) -def _synchronize_textiowrapper(tio, lock): +def _synchronize_textiowrapper(tio: t.TextIO, lock: threading.RLock): # Ensure that a background thread can't hold the internal buffer lock on a file object # during a fork, which causes forked children to hang. We're using display's existing lock for # convenience (and entering the lock before a fork). @@ -207,11 +211,11 @@ def locking_wrapper(*args, **kwargs): buffer = tio.buffer # monkeypatching the underlying file-like object isn't great, but likely safer than subclassing - buffer.write = _wrap_with_lock(buffer.write, lock) - buffer.flush = _wrap_with_lock(buffer.flush, lock) + buffer.write = _wrap_with_lock(buffer.write, lock) # type: ignore[method-assign] + buffer.flush = _wrap_with_lock(buffer.flush, lock) # type: ignore[method-assign] -def setraw(fd, when=termios.TCSAFLUSH): +def setraw(fd: int, when: int = termios.TCSAFLUSH) -> None: """Put terminal into a raw mode. Copied from ``tty`` from CPython 3.11.0, and modified to not remove OPOST from OFLAG @@ -232,13 +236,12 @@ def setraw(fd, when=termios.TCSAFLUSH): termios.tcsetattr(fd, when, mode) -def clear_line(stdout): +def clear_line(stdout: t.BinaryIO) -> None: stdout.write(b'\x1b[%s' % MOVE_TO_BOL) stdout.write(b'\x1b[%s' % CLEAR_TO_EOL) -def setup_prompt(stdin_fd, stdout_fd, seconds, echo): - # type: (int, int, int, bool) -> None +def setup_prompt(stdin_fd: int, stdout_fd: int, seconds: int, echo: bool) -> None: setraw(stdin_fd) # Only set stdout to raw mode if it is a TTY. This is needed when redirecting @@ -252,7 +255,7 @@ def setup_prompt(stdin_fd, stdout_fd, seconds, echo): termios.tcsetattr(stdin_fd, termios.TCSANOW, new_settings) -def setupterm(): +def setupterm() -> None: # Nest the try except since curses.error is not available if curses did not import try: curses.setupterm() @@ -269,9 +272,9 @@ def setupterm(): class Display(metaclass=Singleton): - def __init__(self, verbosity=0): + def __init__(self, verbosity: int = 0) -> None: - self._final_q = None + self._final_q: FinalQueue | None = None # NB: this lock is used to both prevent intermingled output between threads and to block writes during forks. # Do not change the type of this lock or upgrade to a shared lock (eg multiprocessing.RLock). @@ -281,11 +284,11 @@ def __init__(self, verbosity=0): self.verbosity = verbosity # list of all deprecation messages to prevent duplicate display - self._deprecations = {} - self._warns = {} - self._errors = {} + self._deprecations: dict[str, int] = {} + self._warns: dict[str, int] = {} + self._errors: dict[str, int] = {} - self.b_cowsay = None + self.b_cowsay: bytes | None = None self.noncow = C.ANSIBLE_COW_SELECTION self.set_cowsay_info() @@ -296,12 +299,12 @@ def __init__(self, verbosity=0): (out, err) = cmd.communicate() if cmd.returncode: raise Exception - self.cows_available = {to_text(c) for c in out.split()} # set comprehension + self.cows_available: set[str] = {to_text(c) for c in out.split()} if C.ANSIBLE_COW_ACCEPTLIST and any(C.ANSIBLE_COW_ACCEPTLIST): self.cows_available = set(C.ANSIBLE_COW_ACCEPTLIST).intersection(self.cows_available) except Exception: # could not execute cowsay for some reason - self.b_cowsay = False + self.b_cowsay = None self._set_column_width() @@ -321,7 +324,7 @@ def __init__(self, verbosity=0): self.setup_curses = False - def _replacing_warning_handler(self, exception): + def _replacing_warning_handler(self, exception: UnicodeError) -> tuple[str | bytes, int]: # TODO: This should probably be deferred until after the current display is completed # this will require some amount of new functionality self.deprecated( @@ -330,7 +333,7 @@ def _replacing_warning_handler(self, exception): ) return '?', exception.end - def set_queue(self, queue): + def set_queue(self, queue: FinalQueue) -> None: """Set the _final_q on Display, so that we know to proxy display over the queue instead of directly writing to stdout/stderr from forks @@ -340,7 +343,7 @@ def set_queue(self, queue): raise RuntimeError('queue cannot be set in parent process') self._final_q = queue - def set_cowsay_info(self): + def set_cowsay_info(self) -> None: if C.ANSIBLE_NOCOWS: return @@ -352,7 +355,15 @@ def set_cowsay_info(self): self.b_cowsay = b_cow_path @proxy_display - def display(self, msg, color=None, stderr=False, screen_only=False, log_only=False, newline=True): + def display( + self, + msg: str, + color: str | None = None, + stderr: bool = False, + screen_only: bool = False, + log_only: bool = False, + newline: bool = True, + ) -> None: """ Display a message to the user Note: msg *must* be a unicode string to prevent UnicodeError tracebacks. @@ -414,32 +425,32 @@ def display(self, msg, color=None, stderr=False, screen_only=False, log_only=Fal # actually log logger.log(lvl, msg2) - def v(self, msg, host=None): + def v(self, msg: str, host: str | None = None) -> None: return self.verbose(msg, host=host, caplevel=0) - def vv(self, msg, host=None): + def vv(self, msg: str, host: str | None = None) -> None: return self.verbose(msg, host=host, caplevel=1) - def vvv(self, msg, host=None): + def vvv(self, msg: str, host: str | None = None) -> None: return self.verbose(msg, host=host, caplevel=2) - def vvvv(self, msg, host=None): + def vvvv(self, msg: str, host: str | None = None) -> None: return self.verbose(msg, host=host, caplevel=3) - def vvvvv(self, msg, host=None): + def vvvvv(self, msg: str, host: str | None = None) -> None: return self.verbose(msg, host=host, caplevel=4) - def vvvvvv(self, msg, host=None): + def vvvvvv(self, msg: str, host: str | None = None) -> None: return self.verbose(msg, host=host, caplevel=5) - def debug(self, msg, host=None): + def debug(self, msg: str, host: str | None = None) -> None: if C.DEFAULT_DEBUG: if host is None: self.display("%6d %0.5f: %s" % (os.getpid(), time.time(), msg), color=C.COLOR_DEBUG) else: self.display("%6d %0.5f [%s]: %s" % (os.getpid(), time.time(), host, msg), color=C.COLOR_DEBUG) - def verbose(self, msg, host=None, caplevel=2): + def verbose(self, msg: str, host: str | None = None, caplevel: int = 2) -> None: to_stderr = C.VERBOSE_TO_STDERR if self.verbosity > caplevel: @@ -448,7 +459,14 @@ def verbose(self, msg, host=None, caplevel=2): else: self.display("<%s> %s" % (host, msg), color=C.COLOR_VERBOSE, stderr=to_stderr) - def get_deprecation_message(self, msg, version=None, removed=False, date=None, collection_name=None): + def get_deprecation_message( + self, + msg: str, + version: str | None = None, + removed: bool = False, + date: str | None = None, + collection_name: str | None = None, + ) -> str: ''' used to print out a deprecation message.''' msg = msg.strip() if msg and msg[-1] not in ['!', '?', '.']: @@ -484,7 +502,14 @@ def get_deprecation_message(self, msg, version=None, removed=False, date=None, c return message_text @proxy_display - def deprecated(self, msg, version=None, removed=False, date=None, collection_name=None): + def deprecated( + self, + msg: str, + version: str | None = None, + removed: bool = False, + date: str | None = None, + collection_name: str | None = None, + ) -> None: if not removed and not C.DEPRECATION_WARNINGS: return @@ -501,7 +526,7 @@ def deprecated(self, msg, version=None, removed=False, date=None, collection_nam self._deprecations[message_text] = 1 @proxy_display - def warning(self, msg, formatted=False): + def warning(self, msg: str, formatted: bool = False) -> None: if not formatted: new_msg = "[WARNING]: %s" % msg @@ -514,11 +539,11 @@ def warning(self, msg, formatted=False): self.display(new_msg, color=C.COLOR_WARN, stderr=True) self._warns[new_msg] = 1 - def system_warning(self, msg): + def system_warning(self, msg: str) -> None: if C.SYSTEM_WARNINGS: self.warning(msg) - def banner(self, msg, color=None, cows=True): + def banner(self, msg: str, color: str | None = None, cows: bool = True) -> None: ''' Prints a header-looking line with cowsay or stars with length depending on terminal width (3 minimum) ''' @@ -541,7 +566,7 @@ def banner(self, msg, color=None, cows=True): stars = u"*" * star_len self.display(u"\n%s %s" % (msg, stars), color=color) - def banner_cowsay(self, msg, color=None): + def banner_cowsay(self, msg: str, color: str | None = None) -> None: if u": [" in msg: msg = msg.replace(u"[", u"") if msg.endswith(u"]"): @@ -558,7 +583,7 @@ def banner_cowsay(self, msg, color=None): (out, err) = cmd.communicate() self.display(u"%s\n" % to_text(out), color=color) - def error(self, msg, wrap_text=True): + def error(self, msg: str, wrap_text: bool = True) -> None: if wrap_text: new_msg = u"\n[ERROR]: %s" % msg wrapped = textwrap.wrap(new_msg, self.columns) @@ -570,14 +595,24 @@ def error(self, msg, wrap_text=True): self._errors[new_msg] = 1 @staticmethod - def prompt(msg, private=False): + def prompt(msg: str, private: bool = False) -> str: if private: return getpass.getpass(msg) else: return input(msg) - def do_var_prompt(self, varname, private=True, prompt=None, encrypt=None, confirm=False, salt_size=None, salt=None, default=None, unsafe=None): - + def do_var_prompt( + self, + varname: str, + private: bool = True, + prompt: str | None = None, + encrypt: str | None = None, + confirm: bool = False, + salt_size: int | None = None, + salt: str | None = None, + default: str | None = None, + unsafe: bool = False, + ) -> str: result = None if sys.__stdin__.isatty(): @@ -619,14 +654,21 @@ def do_var_prompt(self, varname, private=True, prompt=None, encrypt=None, confir result = wrap_var(result) return result - def _set_column_width(self): + def _set_column_width(self) -> None: if os.isatty(1): tty_size = unpack('HHHH', fcntl.ioctl(1, termios.TIOCGWINSZ, pack('HHHH', 0, 0, 0, 0)))[1] else: tty_size = 0 self.columns = max(79, tty_size - 1) - def prompt_until(self, msg, private=False, seconds=None, interrupt_input=None, complete_input=None): + def prompt_until( + self, + msg: str, + private: bool = False, + seconds: int | None = None, + interrupt_input: c.Container[bytes] | None = None, + complete_input: c.Container[bytes] | None = None, + ) -> bytes: if self._final_q: from ansible.executor.process.worker import current_worker self._final_q.send_prompt( @@ -678,12 +720,11 @@ def prompt_until(self, msg, private=False, seconds=None, interrupt_input=None, c def _read_non_blocking_stdin( self, - echo=False, # type: bool - seconds=None, # type: int - interrupt_input=None, # type: t.Iterable[bytes] - complete_input=None, # type: t.Iterable[bytes] - ): # type: (...) -> bytes - + echo: bool = False, + seconds: int | None = None, + interrupt_input: c.Container[bytes] | None = None, + complete_input: c.Container[bytes] | None = None, + ) -> bytes: if self._final_q: raise NotImplementedError @@ -732,7 +773,7 @@ def _read_non_blocking_stdin( return result_string @property - def _stdin(self): + def _stdin(self) -> t.BinaryIO | None: if self._final_q: raise NotImplementedError try: @@ -741,20 +782,20 @@ def _stdin(self): return None @property - def _stdin_fd(self): + def _stdin_fd(self) -> int | None: try: return self._stdin.fileno() except (ValueError, AttributeError): return None @property - def _stdout(self): + def _stdout(self) -> t.BinaryIO: if self._final_q: raise NotImplementedError return sys.stdout.buffer @property - def _stdout_fd(self): + def _stdout_fd(self) -> int | None: try: return self._stdout.fileno() except (ValueError, AttributeError): diff --git a/lib/ansible/utils/encrypt.py b/lib/ansible/utils/encrypt.py index a0d7c00c9a3bfa..4efcdde429b339 100644 --- a/lib/ansible/utils/encrypt.py +++ b/lib/ansible/utils/encrypt.py @@ -102,7 +102,7 @@ def __init__(self, algorithm): "Python crypt module is deprecated and will be removed from " "Python 3.13. Install the passlib library for continued " "encryption functionality.", - version=2.17 + version="2.17", ) self.algo_data = self.algorithms[algorithm] diff --git a/lib/ansible/vars/plugins.py b/lib/ansible/vars/plugins.py index 9049ebb536d6a6..215f097538159e 100644 --- a/lib/ansible/vars/plugins.py +++ b/lib/ansible/vars/plugins.py @@ -64,7 +64,7 @@ def get_vars_from_path(loader, path, entities, stage): needs_enabled = plugin.REQUIRES_ENABLED elif hasattr(plugin, 'REQUIRES_WHITELIST'): display.deprecated("The VarsModule class variable 'REQUIRES_WHITELIST' is deprecated. " - "Use 'REQUIRES_ENABLED' instead.", version=2.18) + "Use 'REQUIRES_ENABLED' instead.", version="2.18") needs_enabled = plugin.REQUIRES_WHITELIST # A collection plugin was enabled to get to this point because vars_loader.all() does not include collection plugins.