# tbot, Embedded Automation Tool
# Copyright (C) 2019 Harald Seiler
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import base64
import errno
import itertools
import os
import pathlib
import typing
from typing import Any, Generic, Iterable, List, Sequence, Tuple
import tbot.error
from .. import channel # noqa: F401
from .. import linux # noqa: F401
H = typing.TypeVar("H", bound="linux.LinuxShell")
class PathWriteDeathStringException(channel.DeathStringException):
pass
[docs]class Path(typing.Generic[H]):
"""
A path that is associated with a tbot machine.
A path can only be used with its associated host. Using it with
any other host will raise an exception and will be detected by
a static typechecker.
Apart from that, ``Path`` behaves like a :class:`pathlib.Path`::
from tbot.machine import linux
p = linux.Path(mach, "/foo/bar")
p2 = p / "bar" / "baz"
if not p2.exists():
mach.exec0("mkdir", "-p", p2.parent)
mach.exec0("touch", p2)
elif not p2.is_file():
raise Exception(f"{p2} must be a normal file!")
"""
__slots__ = ("_host", "_path")
def __init__(self, host: H, *args: typing.Any) -> None:
"""
Create a new path.
:param linux.LinuxShell host: Host this path should be associated with
:param args: :py:class:`pathlib.PurePosixPath` constructor arguments
.. versionchanged:: 0.9.1
``Path`` no longer inherits from ``pathlib.PurePosixPath``.
"""
self._host: H = host
self._path: pathlib.PurePosixPath = pathlib.PurePosixPath(
*self._prepare_args_list(args)
)
def _prepare_args_list(self, args: Iterable[Any]) -> List[Any]:
ret = list()
for arg in args:
if isinstance(arg, Path):
if arg.host != self.host:
raise tbot.error.WrongHostError(arg, self.host)
ret.append(arg._path)
else:
ret.append(arg)
return ret
# tbot specific API {{{
# These methods are tbot-specific and do not have an equivalent in pathlib.
@property
def host(self) -> H:
"""Host associated with this path."""
return self._host
def _local_str(self) -> str:
return str(self._path)
[docs] def at_host(self, host: H) -> str:
"""
Convert this ``Path`` into a string representation assuming it should
be valid for the machine ``host``. An exception is raised if this
``Path`` is for a different machine instead.
This prevents accidentally using a ``Path`` with the wrong host as much
as possible.
.. versionadded:: 0.9.1
"""
if self.host != host:
raise tbot.error.WrongHostError(self, host)
return str(self._path)
# }}}
# PurePosixPath like API {{{
# Mimick the API of pathlib's PurePosixPath as much as possible. This
# allows users to use tbot's Path just like the one they know from pathlib.
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Path):
return NotImplemented
return self._host == other._host and self._path == other._path
def __hash__(self) -> int:
return hash((self._host, self._path))
def __lt__(self, other: Any) -> bool:
if not isinstance(other, Path):
return NotImplemented
return self._path < other._path
def __le__(self, other: Any) -> bool:
if not isinstance(other, Path):
return NotImplemented
return self._path <= other._path
def __gt__(self, other: Any) -> bool:
if not isinstance(other, Path):
return NotImplemented
return self._path > other._path
def __ge__(self, other: Any) -> bool:
if not isinstance(other, Path):
return NotImplemented
return self._path >= other._path
@property
def name(self) -> str:
return self._path.name
@property
def suffix(self) -> str:
return self._path.suffix
@property
def suffixes(self) -> List[str]:
return self._path.suffixes
@property
def stem(self) -> str:
return self._path.stem
def with_name(self, name: str) -> "Path[H]":
return Path(self._host, self._path.with_name(name))
def with_stem(self, stem: str) -> "Path[H]":
# Not using `self._path.with_stem()` because this only exists in Python 3.9+.
return Path(self._host, self._path.with_name(stem + self._path.suffix))
def with_suffix(self, suffix: str) -> "Path[H]":
return Path(self._host, self._path.with_suffix(suffix))
def relative_to(self, *other: Any) -> "Path[H]":
return Path(self._host, self._path.relative_to(*self._prepare_args_list(other)))
def is_relative_to(self, *other: Any) -> bool:
# Not using `self._path.is_relative_to()` because this only exists in Python 3.9+.
try:
self.relative_to(*other)
return True
except ValueError:
return False
@property
def parts(self) -> Tuple[str, ...]:
return self._path.parts
def joinpath(self, *args: Any) -> "Path[H]":
return Path(self._host, self._path.joinpath(*self._prepare_args_list(args)))
def __truediv__(self, key: Any) -> "Path[H]":
return self.joinpath(key)
def __rtruediv__(self, key: Any) -> "Path[H]":
return Path(self._host, [key] + list(self.parts))
@property
def parent(self) -> "Path[H]":
return Path(self._host, self._path.parent)
@property
def parents(self) -> "_PathParents[H]":
return _PathParents(self)
def is_absolute(self) -> bool:
return self._path.is_absolute()
def match(self, path_pattern: Any) -> bool:
return self._path.match(path_pattern)
# }}}
# Path like API {{{
# Implement some of the methods from pathlib's `Path` class to provide
# convenient access to files on the remote host. All these methods need to
# be implemented by using shell commands on the other side.
[docs] def stat(self) -> os.stat_result:
"""
Return the result of ``stat`` on this path.
Tries to imitate the results of :meth:`pathlib.Path.stat`, returns a
:class:`os.stat_result`.
"""
ec, stat_str = self.host.exec("stat", "-t", self)
if ec != 0:
raise OSError(errno.ENOENT, f"Can't stat {self}")
stat_res = stat_str[len(self.at_host(self.host)) + 1 :].split(" ")
return os.stat_result(
(
int(stat_res[2], 16),
int(stat_res[6]),
0,
int(stat_res[7]),
int(stat_res[3]),
int(stat_res[4]),
int(stat_res[0]),
int(stat_res[10]),
int(stat_res[11]),
int(stat_res[12]),
)
)
[docs] def exists(self) -> bool:
"""Whether this path exists."""
return self.host.test("test", "-e", self)
[docs] def is_dir(self) -> bool:
"""Whether this path points to a directory."""
return self.host.test("test", "-d", self)
[docs] def is_file(self) -> bool:
"""Whether this path points to a normal file."""
return self.host.test("test", "-f", self)
[docs] def is_symlink(self) -> bool:
"""Whether this path points to a symlink."""
return self.host.test("test", "-h", self)
[docs] def is_block_device(self) -> bool:
"""Whether this path points to a block device."""
return self.host.test("test", "-b", self)
[docs] def is_char_device(self) -> bool:
"""Whether this path points to a character device."""
return self.host.test("test", "-c", self)
[docs] def is_fifo(self) -> bool:
"""Whether this path points to a pipe(fifo)."""
return self.host.test("test", "-p", self)
[docs] def is_socket(self) -> bool:
"""Whether this path points to a unix domain-socket."""
return self.host.test("test", "-S", self)
def _glob_inner(
self, path: "Path[H]", *find_args: str
) -> "typing.Iterator[Path[H]]":
output = self.host.exec(
"find",
path,
*find_args,
linux.RedirStderr(Path(self.host, "/dev/null")),
)[1]
for match in output[:-1].split("\n"):
# when no paths matched, we would produce an empty path without
# this short-circuit here:
if match == "":
continue
# filter out this path itself to match `pathlib` behavior
if match != self.at_host(self.host):
yield Path(self.host, match)
[docs] def glob(self, pattern: str) -> "typing.Iterator[Path[H]]":
"""
Iterate over this subtree and yield all existing files (of any
kind, including directories) matching the given relative pattern.
**Example**:
.. code-block:: python
ubootdir = lh.workdir / "u-boot"
# .glob() returns a list which can be iterated.
for f in ubootdir.glob("common/*.c"):
tbot.log.message(f"Found {f}.")
# To use the globs in another commandline (note the `*`!):
lh.exec0("ls", "-l", *ubootdir.glob("common/*.c"))
.. note::
tbot ``Path.glob()``'s behavior is not quite identical to
:py:meth:`pathlib.Path.glob()`. There are two key differences:
1. Only the last path component of the pattern may include globs.
Any globbing in parent components is ignored. Thus, the
following pattern **does not work**: ``co*/*.c``
2. The ``**`` sequence is not supported. In most cases, you can
use :py:meth:`Path.rglob()` instead.
.. versionchanged:: 0.9.6
``Path.glob()`` now properly escapes the pattern so even paths with
spaces are safe. However, globbing is now only supported in the
last component of the path.
"""
# strip path prefix from pattern and "attach" it to the path we're searching
pattern_path = self / pattern
yield from self._glob_inner(
pattern_path.parent, "-maxdepth", "1", "-name", pattern_path.name
)
[docs] def rglob(self, pattern: str) -> "typing.Iterator[Path[H]]":
"""
Recursively match all files beneath this path against ``pattern``.
This is essentially equivalent to ``find <self> -path '*/<pattern>'``.
Note that this path itself will always be excluded from the results.
This method returns an iterator over matching paths.
.. versionadded:: 0.9.6
"""
yield from self._glob_inner(self, "-path", f"*/{pattern}")
[docs] def resolve(self, strict: bool = False) -> "Path[H]":
"""
Make the path absolute, resolving any symlinks. A new path object is returned.
If the path doesn’t exist and ``strict`` is ``True``,
:py:class:`FileNotFoundError` is raised. If ``strict`` is ``False``,
the last component may not exist. The ``False`` behavior slightly
differs from the one in Python's :py:mod:`pathlib`.
.. versionadded:: 0.9.6
"""
resolved = self.host.exec0("realpath", self).strip("\n")
resolved_path = Path(self.host, resolved)
if strict and not resolved_path.exists():
raise FileNotFoundError(resolved_path)
return resolved_path
[docs] def readlink(self) -> "Path[H]":
"""
Return the path to which the symbolic link points.
.. versionadded:: 0.9.6
"""
result = self.host.exec0("readlink", self).strip("\n")
return Path(self.host, result)
[docs] def symlink_to(self, target: "Path[H]") -> None:
"""
Make this path a symbolic link to ``target``. **Example**:
.. code-block:: python
link = host.workdir / "link-name"
target = host.fsroot / "etc" / "os-release"
link.symlink_to(target)
.. versionadded:: 0.9.6
"""
self.host.exec0("ln", "-snf", target, self)
[docs] def write_text(
self,
data: str,
encoding: typing.Optional[str] = None,
errors: typing.Optional[str] = None,
) -> int:
"""
Write ``data`` into the file this path points to.
**Example**:
.. code-block:: python
f = lnx.workdir / "foo.sh"
f.write_text('''\\
#!/bin/sh
set -e
echo "Hello tbot!"
ps ax
''')
f.exec0("chmod", "+x", f)
.. warning::
This string must contain 'text' in the sense that some control
characters are not allowed. Consult the documentation of
:py:meth:`LinuxShell.run() <tbot.machine.linux.LinuxShell.run>` for
details.
Additionally, line-endings might be transformed according to the
tty's settings. This function is not meant for byte-by-byte
transfers, but for configuration files or small scripts. If you
want to transfer a blob or a larger file, consider using
:py:func:`tbot.tc.shell.copy` or (for small files)
:py:meth:`Path.write_bytes() <tbot.machine.linux.Path.write_bytes>`.
"""
if not isinstance(data, str):
raise TypeError(f"data must be str, not {data.__class__.__name__}")
# fast path for single line text. `encoding` and `errors` are ignored
# in this case for now because `bytes` is not a supported argument type.
if "\n" not in data and "\r" not in data:
self.host.exec0("printf", "%s", data, linux.RedirStdout(self))
return len(data)
byte_data = data.encode(encoding or "utf-8", errors or "strict")
with self.host.run(
"tee", self, linux.RedirStdout(self.host.fsroot / "/dev/null")
) as ch:
with ch.with_death_string("tee: ", PathWriteDeathStringException):
try:
ch.send(byte_data, read_back=True)
# Send ^D twice if the file does not end with a line ending. This
# is necessary to make `tee` notice the EOF correctly.
if not (byte_data == b"" or byte_data[-1:] in [b"\n", b"\r"]):
ch.sendcontrol("D")
except PathWriteDeathStringException:
pass
ch.sendcontrol("D")
ch.terminate0()
return len(byte_data)
[docs] def read_text(
self, encoding: typing.Optional[str] = None, errors: typing.Optional[str] = None
) -> str:
"""
Read the contents of a *text* file, pointed to by this path.
.. warning::
This method is for 'text' content only as line-endings might not be
transferred as contained in the file (Will always use a single
``\\n``). If you want to transfer a file byte-by-byte, consider
using :py:func:`tbot.tc.shell.copy` instead. For small files,
:py:meth:`Path.write_bytes() <tbot.machine.linux.Path.read_bytes>`
might also be an option.
"""
if encoding is not None or errors is not None:
raise NotImplementedError("Encoding is not implemented for `read_text`")
return self.host.exec0("cat", self)
[docs] def write_bytes(self, data: bytes) -> int:
"""
Write binary ``data`` into the file this path points to.
.. note::
This method ensures exact byte-by-byte transfer. To do so, it
encodes the data using base64 which makes console output less
readable. If you intend to transfer text data, please use
:py:meth:`Path.write_text() <tbot.machine.linux.Path.write_text>`.
"""
if not isinstance(data, bytes):
raise TypeError(f"data must be bytes, not {data.__class__.__name__}")
with self.host.run(
*["base64", "-d", "-"],
linux.Pipe,
"tee",
self,
linux.RedirStdout(self.host.fsroot / "/dev/null"),
) as ch:
with ch.with_death_string("tee: ", PathWriteDeathStringException):
try:
# Encode as base64 and split into 76 character lines. This makes
# output more readable and does not affect parsing on the remote
# end.
encoded = iter(base64.b64encode(data))
while True:
chunk = bytes(itertools.islice(encoded, 76))
if chunk == b"":
break
ch.sendline(chunk, read_back=True)
except PathWriteDeathStringException:
pass
ch.sendcontrol("D")
ch.terminate0()
return len(data)
[docs] def read_bytes(self) -> bytes:
"""
Read the contents of a file, pointed to by this path.
.. note::
This method ensures exact byte-by-byte transfer. To do so, it
encodes the data using base64 which makes console output less
readable. If you intend to transfer text data, please use
:py:meth:`Path.read_text() <tbot.machine.linux.Path.read_text>`.
"""
encoded = self.host.exec0("base64", self)
return base64.b64decode(encoded)
[docs] def rmdir(self) -> None:
"""
Remove the directory pointed to by this path. The directory must be
empty.
.. versionadded:: 0.9.1
"""
if self.is_symlink() or not self.is_dir():
raise NotADirectoryError
self.host.exec0("rmdir", self)
[docs] def unlink(self, missing_ok: bool = False) -> None:
"""
Remove the file or symbolic link from the filesystem. If the path points
to a directory, use :py:meth:`Path.rmdir()
<tbot.machine.linux.Path.rmdir>` instead.
If ``missing_ok`` is false (the default), ``FileNotFoundError`` is
raised if the path does not exist.
.. versionadded:: 0.9.1
"""
if not self.exists():
if not missing_ok:
raise FileNotFoundError(
errno.ENOENT, os.strerror(errno.ENOENT), str(self)
)
return
self.host.exec0("rm", self)
[docs] def mkdir(self, parents: bool = False, exist_ok: bool = False) -> None:
"""
Create a directory at the path this object represents.
If ``parents`` is false (the default), a missing parent causes
``FileNotFoundError`` to be raised. If ``parents`` is true, missing
parent directories are created as needed.
If ``exist_ok`` is false (the default), a ``FileExistsError`` is raised
if the path already exists. If ``exist_ok`` is true, a
``FileExistsError`` exception be suppressed if the target path exists
and is a directory.
.. versionadded:: 0.9.1
"""
if not self.parent.exists() and not parents:
raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), str(self))
if self.exists():
if not exist_ok or not self.is_dir():
raise FileExistsError(
errno.EEXIST, os.strerror(errno.EEXIST), str(self)
)
return
if parents:
self.host.exec0("mkdir", "-p", self)
else:
self.host.exec0("mkdir", self)
# }}}
def __str__(self) -> str:
return f"{self._host.name}:{self._path}"
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self._host!r}, {str(self._path)!r})"
class _PathParents(Sequence[Path[H]], Generic[H]):
"""
This object provides sequence-like access to the logical ancestors
of a path. Don't try to construct it yourself.
"""
__slots__ = ("_pathcls", "_host", "_parts")
def __init__(self, path: Path[H]):
self._pathcls = type(path)
self._host = path.host
self._parts = path.parts
def __len__(self) -> int:
return len(self._parts)
def __getitem__(self, idx: int) -> Path[H]: # type: ignore
if idx < 0 or idx >= len(self):
raise IndexError(idx)
return self._pathcls(self._host, *self._parts[: -idx - 1])
def __repr__(self) -> str:
return f"<{self._pathcls.__name__}.parents>"
