Drop chroot setting possibility in __init__ and public property

Design issue

Signed-off-by: Aleksei Stepanov <penguinolog@gmail.com>

(cherry picked from commit 52aa1b3)
Signed-off-by: Aleksei Stepanov <penguinolog@gmail.com>

Author: penguinolog

doc/source/SSHClient.rst

@@ -10,7 +10,7 @@ API: SSHClient and SSHAuth.
 
     SSHClient helper.
 
-    .. py:method:: __init__(host, port=22, username=None, password=None, private_keys=None, auth=None, *, chroot_path=None)
+    .. py:method:: __init__(host, port=22, username=None, password=None, private_keys=None, auth=None)
 
         :param host: remote hostname
         :type host: ``str``
@@ -26,8 +26,6 @@ API: SSHClient and SSHAuth.
         :type auth: typing.Optional[SSHAuth]
         :param verbose: show additional error/warning messages
         :type verbose: bool
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
 
     .. note:: auth has priority over username/password/private_keys
 
@@ -68,11 +66,6 @@ API: SSHClient and SSHAuth.
         ``bool``
         Paramiko status: ready to use|reconnect required
 
-    .. py:attribute:: chroot_path
-
-        ``typing.Optional[str]``
-        Path for chroot if set.
-
     .. py:attribute:: sudo_mode
 
         ``bool``

doc/source/Subprocess.rst

@@ -8,16 +8,14 @@ API: Subprocess
 
 .. py:class:: Subprocess()
 
-    .. py:method:: __init__(logger, log_mask_re=None, *, logger=logging.getLogger("exec_helpers.subprocess_runner"), chroot_path=None)
+    .. py:method:: __init__(logger, log_mask_re=None, *, logger=logging.getLogger("exec_helpers.subprocess_runner"))
 
         ExecHelper global API.
 
         :param log_mask_re: regex lookup rule to mask command for logger. all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
         :param logger: logger instance to use
         :type logger: logging.Logger
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         .. versionchanged:: 3.1.0 Not singleton anymore. Only lock is shared between all instances.
@@ -46,11 +44,6 @@ API: Subprocess
 
         .. versionchanged:: 1.1.0 release lock on exit
 
-    .. py:attribute:: chroot_path
-
-        ``typing.Optional[str]``
-        Path for chroot if set.
-
     .. py:method:: chroot(path)
 
         Context manager for changing chroot rules.

exec_helpers/_ssh_client_base.py

@@ -138,9 +138,7 @@ def __call__(  # type: ignore
         password: typing.Optional[str] = None,
         private_keys: typing.Optional[typing.Iterable[paramiko.RSAKey]] = None,
         auth: typing.Optional[ssh_auth.SSHAuth] = None,
-        verbose: bool = True,
-        *,
-        chroot_path: typing.Optional[str] = None
+        verbose: bool = True
     ) -> "SSHClientBase":
         """Main memorize method: check for cached instance and return it. API follows target __init__.
 
@@ -158,12 +156,10 @@ def __call__(  # type: ignore
         :type auth: typing.Optional[ssh_auth.SSHAuth]
         :param verbose: show additional error/warning messages
         :type verbose: bool
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
         :return: SSH client instance
         :rtype: SSHClientBase
         """
-        if (host, port) in cls.__cache and not chroot_path:  # chrooted connections are not memorized
+        if (host, port) in cls.__cache:
             key = host, port
             if auth is None:
                 auth = ssh_auth.SSHAuth(username=username, password=password, keys=private_keys)
@@ -191,10 +187,8 @@ def __call__(  # type: ignore
             private_keys=private_keys,
             auth=auth,
             verbose=verbose,
-            chroot_path=chroot_path,
         )
-        if not chroot_path:
-            cls.__cache[(ssh.hostname, ssh.port)] = ssh
+        cls.__cache[(ssh.hostname, ssh.port)] = ssh
         return ssh
 
     @classmethod
@@ -292,9 +286,7 @@ def __init__(
         password: typing.Optional[str] = None,
         private_keys: typing.Optional[typing.Iterable[paramiko.RSAKey]] = None,
         auth: typing.Optional[ssh_auth.SSHAuth] = None,
-        verbose: bool = True,
-        *,
-        chroot_path: typing.Optional[str] = None
+        verbose: bool = True
     ) -> None:
         """Main SSH Client helper.
 
@@ -312,14 +304,11 @@ def __init__(
         :type auth: typing.Optional[ssh_auth.SSHAuth]
         :param verbose: show additional error/warning messages
         :type verbose: bool
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
 
         .. note:: auth has priority over username/password/private_keys
         """
         super(SSHClientBase, self).__init__(
-            logger=logging.getLogger(self.__class__.__name__).getChild("{host}:{port}".format(host=host, port=port)),
-            chroot_path=chroot_path
+            logger=logging.getLogger(self.__class__.__name__).getChild("{host}:{port}".format(host=host, port=port))
         )
 
         self.__hostname = host

exec_helpers/api.py

@@ -46,13 +46,14 @@
 )
 
 
-class _ChRootContext:
+# noinspection PyProtectedMember
+class _ChRootContext:  # pylint: disable=protected-access
     """Context manager for call commands with chroot.
 
     .. versionadded:: 4.1.0
     """
 
-    __slots__ = ("__conn", "__chroot_status", "__path")
+    __slots__ = ("_conn", "_chroot_status", "_path")
 
     def __init__(self, conn: "ExecHelper", path: typing.Optional[str] = None) -> None:
         """Context manager for call commands with sudo.
@@ -62,41 +63,33 @@ def __init__(self, conn: "ExecHelper", path: typing.Optional[str] = None) -> Non
         :param path: chroot path or None for no chroot
         :type path: typing.Optional[str]
         """
-        self.__conn = conn  # type: ExecHelper
-        self.__chroot_status = conn.chroot_path  # type: typing.Optional[str]
-        self.__path = path  # type: typing.Optional[str]
+        self._conn = conn  # type: ExecHelper
+        self._chroot_status = conn._chroot_path  # type: typing.Optional[str]
+        self._path = path  # type: typing.Optional[str]
 
     def __enter__(self) -> None:
-        self.__conn.__enter__()
-        self.__chroot_status = self.__conn.chroot_path
-        self.__conn.chroot_path = self.__path
+        self._conn.__enter__()
+        self._chroot_status = self._conn._chroot_path
+        self._conn._chroot_path = self._path
 
     def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
-        self.__conn.chroot_path = self.__chroot_status
-        self.__conn.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)  # type: ignore
+        self._conn._chroot_path = self._chroot_status
+        self._conn.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)  # type: ignore
 
 
 class ExecHelper(metaclass=abc.ABCMeta):
     """ExecHelper global API."""
 
     __slots__ = ("__lock", "__logger", "log_mask_re", "__chroot_path")
 
-    def __init__(
-        self,
-        log_mask_re: typing.Optional[str] = None,
-        *,
-        logger: logging.Logger,
-        chroot_path: typing.Optional[str] = None
-    ) -> None:
+    def __init__(self, log_mask_re: typing.Optional[str] = None, *, logger: logging.Logger) -> None:
         """Global ExecHelper API.
 
         :param logger: logger instance to use
         :type logger: logging.Logger
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         .. versionchanged:: 1.3.5 make API public to use as interface
@@ -105,7 +98,7 @@ def __init__(
         self.__lock = threading.RLock()
         self.__logger = logger
         self.log_mask_re = log_mask_re
-        self.__chroot_path = chroot_path  # type: typing.Optional[str]
+        self.__chroot_path = None  # type: typing.Optional[str]
 
     @property
     def logger(self) -> logging.Logger:
@@ -121,16 +114,16 @@ def lock(self) -> threading.RLock:
         return self.__lock
 
     @property
-    def chroot_path(self) -> typing.Optional[str]:
+    def _chroot_path(self) -> typing.Optional[str]:
         """Path for chroot if set.
 
         :rtype: typing.Optional[str]
         .. versionadded:: 3.5.3
         """
         return self.__chroot_path
 
-    @chroot_path.setter
-    def chroot_path(self, new_state: typing.Optional[str]) -> None:
+    @_chroot_path.setter
+    def _chroot_path(self, new_state: typing.Optional[str]) -> None:
         """Path for chroot if set.
 
         :param new_state: new path
@@ -139,8 +132,8 @@ def chroot_path(self, new_state: typing.Optional[str]) -> None:
         """
         self.__chroot_path = new_state
 
-    @chroot_path.deleter
-    def chroot_path(self) -> None:
+    @_chroot_path.deleter
+    def _chroot_path(self) -> None:
         """Remove Path for chroot.
 
         .. versionadded:: 3.5.3
@@ -218,9 +211,9 @@ def mask(text: str, rules: str) -> str:
 
     def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -> str:
         """Prepare command: cower chroot and other cases."""
-        if any((chroot_path, self.chroot_path)):
+        if any((chroot_path, self._chroot_path)):
             return "chroot {chroot_path} {cmd}".format(
-                chroot_path=chroot_path if chroot_path else self.chroot_path,
+                chroot_path=chroot_path if chroot_path else self._chroot_path,
                 cmd=cmd
             )
         return cmd

exec_helpers/async_api/api.py

@@ -31,29 +31,47 @@
 from exec_helpers import proc_enums
 
 
+# noinspection PyProtectedMember
+class _ChRootContext(api._ChRootContext):  # pylint: disable=protected-access
+    """Async extension for chroot."""
+
+    def __init__(self, conn: "ExecHelper", path: typing.Optional[str] = None) -> None:
+        """Context manager for call commands with sudo.
+
+        :param conn: connection instance
+        :type conn: ExecHelper
+        :param path: chroot path or None for no chroot
+        :type path: typing.Optional[str]
+        """
+        super(_ChRootContext, self).__init__(conn=conn, path=path)
+
+    async def __aenter__(self) -> None:
+        await self._conn.__aenter__()  # type: ignore
+        self._chroot_status = self._conn._chroot_path  # pylint: disable=protected-access
+        self._conn._chroot_path = self._path  # pylint: disable=protected-access
+
+    async def __aexit__(  # pylint: disable=protected-access
+        self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any
+    ) -> None:
+        self._conn._chroot_path = self._chroot_status
+        await self._conn.__aexit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)  # type: ignore
+
+
 class ExecHelper(api.ExecHelper, metaclass=abc.ABCMeta):
     """Subprocess helper with timeouts and lock-free FIFO."""
 
     __slots__ = ("__alock",)
 
-    def __init__(
-        self,
-        log_mask_re: typing.Optional[str] = None,
-        *,
-        logger: logging.Logger,
-        chroot_path: typing.Optional[str] = None
-    ) -> None:
+    def __init__(self, log_mask_re: typing.Optional[str] = None, *, logger: logging.Logger) -> None:
         """Subprocess helper with timeouts and lock-free FIFO.
 
         :param logger: logger instance to use
         :type logger: logging.Logger
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
         """
-        super(ExecHelper, self).__init__(logger=logger, log_mask_re=log_mask_re, chroot_path=chroot_path)
+        super(ExecHelper, self).__init__(logger=logger, log_mask_re=log_mask_re)
         self.__alock = None  # type: typing.Optional[asyncio.Lock]
 
     async def __aenter__(self) -> "ExecHelper":
@@ -67,6 +85,19 @@ async def __aexit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typ
         """Async context manager."""
         self.__alock.release()  # type: ignore
 
+    def chroot(self, path: typing.Union[str, None]) -> "typing.ContextManager[None]":
+        """Context manager for changing chroot rules.
+
+        :param path: chroot path or none for working without chroot.
+        :type path: typing.Optional[str]
+        :return: context manager with selected chroot state inside
+        :rtype: typing.ContextManager
+
+        .. Note:: Enter and exit main context manager is produced as well.
+        .. versionadded:: 4.1.0
+        """
+        return _ChRootContext(conn=self, path=path)
+
     @abc.abstractmethod
     async def _exec_command(  # type: ignore
         self,

exec_helpers/async_api/subprocess_runner.py

@@ -71,8 +71,7 @@ def __init__(
         self,
         log_mask_re: typing.Optional[str] = None,
         *,
-        logger: logging.Logger = logging.getLogger(__name__),  # noqa: B008
-        chroot_path: typing.Optional[str] = None
+        logger: logging.Logger = logging.getLogger(__name__)  # noqa: B008
     ) -> None:
         """Subprocess helper with timeouts and lock-free FIFO.
 
@@ -81,13 +80,11 @@ def __init__(
         :type log_mask_re: typing.Optional[str]
         :param logger: logger instance to use
         :type logger: logging.Logger
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 3.1.0 Not singleton anymore. Only lock is shared between all instances.
         .. versionchanged:: 3.2.0 Logger can be enforced.
         """
-        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re, chroot_path=chroot_path)
+        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re)
 
     async def _exec_command(  # type: ignore
         self,

exec_helpers/subprocess_runner.py

@@ -67,8 +67,7 @@ def __init__(
         self,
         log_mask_re: typing.Optional[str] = None,
         *,
-        logger: logging.Logger = logging.getLogger(__name__),  # noqa: B008
-        chroot_path: typing.Optional[str] = None
+        logger: logging.Logger = logging.getLogger(__name__)  # noqa: B008
     ) -> None:
         """Subprocess helper with timeouts and lock-free FIFO.
 
@@ -79,14 +78,12 @@ def __init__(
         :type log_mask_re: typing.Optional[str]
         :param logger: logger instance to use
         :type logger: logging.Logger
-        :param chroot_path: chroot path (use chroot if set)
-        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         .. versionchanged:: 3.1.0 Not singleton anymore. Only lock is shared between all instances.
         .. versionchanged:: 3.2.0 Logger can be enforced.
         """
-        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re, chroot_path=chroot_path)
+        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re)
 
     def _exec_command(  # type: ignore
         self,