Allow pathlib.Path as path for chroot

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

Author: penguinolog

doc/source/SSHClient.rst

@@ -103,7 +103,7 @@ API: SSHClient and SSHAuth.
         Context manager for changing chroot rules.
 
         :param path: chroot path or none for working without chroot.
-        :type path: typing.Optional[str]
+        :type path: typing.Optional[typing.Union[str, pathlib.Path]]
         :return: context manager with selected chroot state inside
         :rtype: typing.ContextManager
 

doc/source/Subprocess.rst

@@ -50,7 +50,7 @@ API: Subprocess
         Context manager for changing chroot rules.
 
         :param path: chroot path or none for working without chroot.
-        :type path: typing.Optional[str]
+        :type path: typing.Optional[typing.Union[str, pathlib.Path]]
         :return: context manager with selected chroot state inside
         :rtype: typing.ContextManager
 

exec_helpers/api.py

@@ -25,6 +25,7 @@
 import abc
 import datetime
 import logging
+import pathlib
 import re
 import shlex
 import threading
@@ -58,17 +59,23 @@ class _ChRootContext:
 
     __slots__ = ("_conn", "_chroot_status", "_path")
 
-    def __init__(self, conn: "ExecHelper", path: typing.Optional[str] = None) -> None:
+    def __init__(self, conn: "ExecHelper", path: typing.Optional[typing.Union[str, pathlib.Path]] = 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]
+        :type path: typing.Optional[typing.Union[str, pathlib.Path]]
+        :raises TypeError: incorrect type of path variable
         """
         self._conn: "ExecHelper" = conn
         self._chroot_status: typing.Optional[str] = conn._chroot_path
-        self._path: typing.Optional[str] = path
+        if path is None or isinstance(path, str):
+            self._path: typing.Optional[str] = path
+        elif isinstance(path, pathlib.Path):
+            self._path = path.as_posix()  # get absolute path
+        else:
+            raise TypeError(f"path={path!r} is not instance of Optional[Union[str, pathlib.Path]]")
 
     def __enter__(self) -> None:
         self._conn.__enter__()
@@ -146,11 +153,11 @@ def _chroot_path(self) -> None:
         """
         self.__chroot_path = None
 
-    def chroot(self, path: typing.Union[str, None]) -> "typing.ContextManager[None]":
+    def chroot(self, path: typing.Union[str, pathlib.Path, 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]
+        :type path: typing.Optional[typing.Union[str, pathlib.Path]]
         :return: context manager with selected chroot state inside
         :rtype: typing.ContextManager
 
@@ -233,6 +240,7 @@ def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -
             return f'chroot {target_path} sh -c {shlex.quote(f"eval {quoted_command}")}'
         return cmd
 
+    # noinspection PyIncorrectDocstring
     def execute_async(  # pylint: disable=missing-param-doc,differing-param-doc,differing-type-doc
         self, *args: typing.Any, **kwargs: typing.Any
     ) -> ExecuteAsyncResult:

exec_helpers/async_api/api.py

@@ -23,6 +23,7 @@
 import abc
 import asyncio
 import logging
+import pathlib
 import typing
 
 # Exec-Helpers Implementation
@@ -37,13 +38,13 @@
 class _ChRootContext(api._ChRootContext):  # pylint: disable=protected-access
     """Async extension for chroot."""
 
-    def __init__(self, conn: "ExecHelper", path: typing.Optional[str] = None) -> None:
+    def __init__(self, conn: "ExecHelper", path: typing.Optional[typing.Union[str, pathlib.Path]] = 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]
+        :type path: typing.Optional[typing.Union[str, pathlib.Path]]
         """
         super(_ChRootContext, self).__init__(conn=conn, path=path)
 
@@ -85,11 +86,11 @@ 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]":
+    def chroot(self, path: typing.Union[str, pathlib.Path, 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]
+        :type path: typing.Optional[typing.Union[str, pathlib.Path]]
         :return: context manager with selected chroot state inside
         :rtype: typing.ContextManager
 

test/test_ssh_client_execute_async_special.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 # Standard Library
+import pathlib
 import shlex
 import typing
 from unittest import mock
@@ -263,3 +264,23 @@ def test_013_execute_async_no_chroot_context(ssh, ssh_transport_channel):
             mock.call.exec_command(f'{command}\n'),
         )
     )
+
+
+def test_012_execute_async_chroot_path(ssh, ssh_transport_channel):
+    """Command-only chroot path."""
+    with ssh.chroot(pathlib.Path('/')):
+        ssh._execute_async(command)
+    ssh_transport_channel.assert_has_calls(
+        (
+            mock.call.makefile_stderr("rb"),
+            mock.call.exec_command(f'chroot / sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
+        )
+    )
+
+
+def test_013_execute_async_chroot_path_invalid_type(ssh, ssh_transport_channel):
+    """Command-only chroot path."""
+    with pytest.raises(TypeError) as exc:
+        with ssh.chroot(...):
+            ssh._execute_async(command)
+        assert str(exc.value) == f"path={...!r} is not instance of Optional[Union[str, pathlib.Path]]"