Align ssh username for logger with paramiko

* Fix code duplication & several SonarLint warnings

Author: penguinolog

exec_helpers/_ssh_client_base.py

@@ -24,11 +24,9 @@
 import datetime
 import getpass
 import logging
-import os
 import shlex
 import socket
 import stat
-import sys
 import time
 import typing
 import warnings
@@ -238,7 +236,7 @@ def __init__(
             for proxy connection auth information is collected from SSHConfig
             if ssh_auth_map record is not available
 
-        .. versionchanged:: 6.0.0 private_keys, auth and vebose became keyword-only arguments
+        .. versionchanged:: 6.0.0 private_keys, auth and verbose became keyword-only arguments
         .. versionchanged:: 6.0.0 added optional ssh_config for ssh-proxy & low level connection parameters handling
         .. versionchanged:: 6.0.0 added optional ssh_auth_map for ssh proxy cases with authentication on each step
         .. versionchanged:: 6.0.0 added optional sock for manual proxy chain handling
@@ -263,7 +261,10 @@ def __init__(
 
         # Save resolved hostname and port
         self.__hostname: str = config.hostname
-        self.__port: int = port if port is not None else config.port if config.port is not None else 22
+        if port is not None:
+            self.__port: int = port
+        else:
+            self.__port = config.port if config.port is not None else 22
 
         # Store initial auth mapping
         self.__auth_mapping = ssh_auth.SSHAuthMapping(ssh_auth_map)
@@ -295,11 +296,11 @@ def __init__(
 
         # Init super with host and real port and username
         mod_name = "exec_helpers" if self.__module__.startswith("exec_helpers") else self.__module__
-        log_username: typing.Optional[str] = self.__get_user_for_log(real_auth)
+        log_username: str = real_auth.username if real_auth.username is not None else getpass.getuser()
 
-        super(SSHClientBase, self).__init__(
+        super().__init__(
             logger=logging.getLogger(f"{mod_name}.{self.__class__.__name__}").getChild(
-                f"({log_username}@{host}:{self.__port})"
+                f"({log_username}@{host}:{self.port})"
             )
         )
 
@@ -315,21 +316,6 @@ def __init__(
 
         self.__connect()
 
-    @staticmethod
-    def __get_user_for_log(real_auth: ssh_auth.SSHAuth) -> typing.Optional[str]:  # pragma: no cover
-        if real_auth.username is not None:
-            return real_auth.username
-        # noinspection PyBroadException
-        try:
-            if sys.platform != "win32":
-                import pwd  # pylint: disable=import-outside-toplevel,import-error
-
-                uid: int = os.getuid()  # pylint: disable=no-member
-                return pwd.getpwuid(uid).pw_name  # Correct for not windows only
-            return getpass.getuser()
-        except Exception:
-            return None
-
     def __rebuild_ssh_config(self) -> None:
         """Rebuild main ssh config from available information."""
         self.__ssh_config[self.hostname] = self.__ssh_config[self.hostname].overridden_by(
@@ -1116,7 +1102,10 @@ def _get_proxy_channel(self, port: typing.Optional[int], ssh_config: SSHConfig,)
 
         .. versionadded:: 6.0.0
         """
-        dest_port: int = port if port is not None else ssh_config.port if ssh_config.port is not None else 22
+        if port is not None:
+            dest_port: int = port
+        else:
+            dest_port = ssh_config.port if ssh_config.port is not None else 22
 
         return self._ssh.get_transport().open_channel(
             kind="direct-tcpip", dest_addr=(ssh_config.hostname, dest_port), src_addr=(self.hostname, 0)
@@ -1257,8 +1246,7 @@ def execute_through_host(
 
         if target_port is not None:  # pragma: no cover
             warnings.warn(
-                f'"target_port" argument was renamed to "port". '
-                f"Old version will be droppped in the next major release.",
+                f'"target_port" argument was renamed to "port". Old version will be dropped in the next major release.',
                 DeprecationWarning,
             )
             port = target_port
@@ -1343,13 +1331,8 @@ def get_result(remote: "SSHClientBase") -> exec_result.ExecResult:
             """
             # pylint: disable=protected-access
             cmd_for_log: str = remote._mask_command(cmd=command, log_mask_re=log_mask_re)
+            remote._log_command_execute(command=command, log_mask_re=log_mask_re, log_level=log_level, **kwargs)
 
-            target_path: typing.Optional[str] = kwargs.get("chroot_path", remote._chroot_path)
-            chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
-
-            remote.logger.log(
-                level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n",
-            )
             async_result: SshExecuteAsyncResult = remote._execute_async(
                 command,
                 stdin=stdin,

exec_helpers/api.py

@@ -50,6 +50,7 @@
 _OptionalTimeoutT = typing.Union[int, float, None]
 _OptionalStdinT = typing.Union[bytes, str, bytearray, None]
 _ExitCodeT = typing.Union[int, proc_enums.ExitCodes]
+_CalledProcessErrorSubClass = typing.Type[exceptions.CalledProcessError]
 
 
 # noinspection PyProtectedMember
@@ -323,6 +324,21 @@ def _exec_command(
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         """
 
+    def _log_command_execute(
+        self,
+        command: str,
+        log_mask_re: typing.Union[str, None],
+        log_level: int,
+        chroot_path: typing.Union[str, None],
+        **_: typing.Any,
+    ) -> None:
+        """Log command execution."""
+        cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
+        target_path: typing.Optional[str] = chroot_path if chroot_path is not None else self._chroot_path
+        chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
+
+        self.logger.log(level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n")
+
     def execute(
         self,
         command: str,
@@ -361,16 +377,8 @@ def execute(
         .. versionchanged:: 1.2.0 default timeout 1 hour
         .. versionchanged:: 2.1.0 Allow parallel calls
         """
-        cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
         log_level: int = logging.INFO if verbose else logging.DEBUG
-
-        target_path: typing.Optional[str] = kwargs.get("chroot_path", self._chroot_path)
-        chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
-
-        self.logger.log(
-            level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n",
-        )
-
+        self._log_command_execute(command=command, log_mask_re=log_mask_re, log_level=log_level, **kwargs)
         async_result: ExecuteAsyncResult = self._execute_async(
             command,
             verbose=verbose,
@@ -454,7 +462,7 @@ def check_call(
         stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
         open_stderr: bool = True,
-        exception_class: "typing.Type[exceptions.CalledProcessError]" = exceptions.CalledProcessError,
+        exception_class: _CalledProcessErrorSubClass = exceptions.CalledProcessError,
         **kwargs: typing.Any,
     ) -> exec_result.ExecResult:
         """Execute command and check for return code.
@@ -528,7 +536,7 @@ def check_stderr(
         stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
         open_stderr: bool = True,
-        exception_class: "typing.Type[exceptions.CalledProcessError]" = exceptions.CalledProcessError,
+        exception_class: _CalledProcessErrorSubClass = exceptions.CalledProcessError,
         **kwargs: typing.Any,
     ) -> exec_result.ExecResult:
         """Execute command expecting return code 0 and empty STDERR.
@@ -595,7 +603,7 @@ def _handle_stderr(
         error_info: typing.Optional[str],
         raise_on_err: bool,
         expected: typing.Iterable[_ExitCodeT],
-        exception_class: "typing.Type[exceptions.CalledProcessError]",
+        exception_class: _CalledProcessErrorSubClass,
     ) -> exec_result.ExecResult:
         """Internal check_stderr logic (synchronous)."""
         append: str = error_info + "\n" if error_info else ""

exec_helpers/async_api/api.py

@@ -219,15 +219,8 @@ async def execute(  # type: ignore
         :rtype: ExecResult
         :raises ExecHelperTimeoutError: Timeout exceeded
         """
-        cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
         log_level: int = logging.INFO if verbose else logging.DEBUG
-
-        target_path: typing.Optional[str] = kwargs.get("chroot_path", self._chroot_path)
-        chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
-
-        self.logger.log(
-            level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n",
-        )
+        self._log_command_execute(command=command, log_mask_re=log_mask_re, log_level=log_level, **kwargs)
 
         async_result: api.ExecuteAsyncResult = await self._execute_async(
             command,

exec_helpers/async_api/subprocess_runner.py

@@ -61,7 +61,7 @@ def interface(self) -> asyncio.subprocess.Process:  # type: ignore
     @property
     def stdin(self) -> typing.Optional[asyncio.StreamWriter]:  # type: ignore
         """Override original NamedTuple with proper typing."""
-        return super(SubprocessExecuteAsyncResult, self).stdin  # type: ignore
+        return super().stdin  # type: ignore
 
     @property
     def stderr(self) -> typing.Optional[typing.AsyncIterable[bytes]]:  # type: ignore
@@ -273,7 +273,7 @@ async def _execute_async(  # type: ignore  # pylint: disable=arguments-differ
                 process.stdin.close()  # type: ignore
             except OSError as exc:
                 if exc.errno in (errno.EINVAL, errno.EPIPE, errno.ESHUTDOWN):
-                    pass
+                    pass  # PIPE already closed
                 else:
                     process.kill()
                     raise

exec_helpers/exceptions.py

@@ -142,7 +142,7 @@ def __init__(self, result: "exec_result.ExecResult", timeout: typing.Union[int,
         :type timeout: typing.Union[int, float]
         """
         message: str = _log_templates.CMD_WAIT_ERROR.format(result=result, timeout=timeout)
-        super(ExecHelperTimeoutError, self).__init__(message, result=result, timeout=timeout)
+        super().__init__(message, result=result, timeout=timeout)
 
 
 class CalledProcessError(ExecCalledProcessError):

exec_helpers/exec_result.py

@@ -556,6 +556,7 @@ def stdout_yaml(self) -> typing.Any:
         with self.stdout_lock:
             return self.__deserialize(fmt="yaml")
 
+    # noinspection PyUnresolvedReferences
     @property
     def stdout_xml(self) -> "xml.etree.ElementTree.Element":
         """XML from stdout.

exec_helpers/ssh_client.py

@@ -115,7 +115,7 @@ def download(self, destination: str, target: str) -> bool:
             if self.exists(destination):
                 self._sftp.get(destination, target)
             else:
-                self.logger.debug(f"Can't download {destination} because it doesn't exist")
+                self.logger.debug(f"Can't download {destination} because it does not exist")
         else:
             self.logger.debug(f"Can't download {destination} because it is a directory")
         return os.path.exists(target)

exec_helpers/subprocess_runner.py

@@ -47,6 +47,7 @@
 _OptionalTimeoutT = typing.Union[int, float, None]
 _OptionalStdinT = typing.Union[bytes, str, bytearray, None]
 _ExitCodeT = typing.Union[int, proc_enums.ExitCodes]
+_OptionalIOBytes = typing.Optional[typing.IO[bytes]]
 
 
 # noinspection PyTypeHints
@@ -59,17 +60,17 @@ def interface(self) -> "subprocess.Popen[bytes]":
         return super().interface  # type: ignore
 
     @property
-    def stdin(self) -> "typing.Optional[typing.IO[bytes]]":  # type: ignore
+    def stdin(self) -> _OptionalIOBytes:  # type: ignore
         """Override original NamedTuple with proper typing."""
-        return super(SubprocessExecuteAsyncResult, self).stdin
+        return super().stdin
 
     @property
-    def stderr(self) -> "typing.Optional[typing.IO[bytes]]":  # type: ignore
+    def stderr(self) -> _OptionalIOBytes:  # type: ignore
         """Override original NamedTuple with proper typing."""
         return super().stderr
 
     @property
-    def stdout(self) -> "typing.Optional[typing.IO[bytes]]":  # type: ignore
+    def stdout(self) -> _OptionalIOBytes:  # type: ignore
         """Override original NamedTuple with proper typing."""
         return super().stdout
 
@@ -287,7 +288,7 @@ def _execute_async(  # pylint: disable=arguments-differ
                 process.stdin.close()
             except OSError as exc:
                 if exc.errno in (errno.EINVAL, errno.EPIPE, errno.ESHUTDOWN):
-                    pass
+                    pass  # PIPE already closed
                 else:
                     process.kill()
                     raise