tomerfiliba / plumbum / 872

pending completion

Build # 872

Build Type

push

travis-ci

Committed by

henryiii

Commit Message

<a href="https://github.com/tomerfiliba/plumbum/commit/<a class=hub.com/tomerfiliba/plumbum/commit/baa953464a5532848671981ee0cceb6fd0d42315">baa953464<a href="https://github.com/tomerfiliba/plumbum/commit/baa953464a5532848671981ee0cceb6fd0d42315">&quot;&gt;Fix #293: AttributeError: &amp;#39;SshMachine&amp;#39; object has no attribute &amp;#39;env&amp;#39; (#347)

* Allow _setdoc() to fetch docstring from a function with different name

* Make BaseRemoteMachine.path() not depend on env

Closes #293 AttributeError: &amp;#39;SshMachine&amp;#39; object has no attribute &amp;#39;env&amp;#39;

* Revert &amp;quot;Allow _setdoc() to fetch docstring from a function with different name&quot;

This reverts commit </a><a class="double-link" href="https://github.com/tomerfiliba/plumbum/commit/<a class="double-link" href="https://github.com/tomerfiliba/plumbum/commit/1b410588970098fcfbc2a8b0c0fa7e71f1cfd637">1b4105889</a>">1b4105889</a><a href="https://github.com/tomerfiliba/plumbum/commit/baa953464a5532848671981ee0cceb6fd0d42315">.

* Share docstrings from BaseRemoteMachine to RemoteEnv.expand*()

* Fix docstring assignment problem with Py2.X

Run Details

1060 of 1570 branches covered (67.52%)

9 of 9 new or added lines in 1 file covered. (100.0%)

3671 of 4459 relevant lines covered (82.33%)

6.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.85

/plumbum/machines/remote.py

import re
from contextlib import contextmanager
from plumbum.commands import CommandNotFound, shquote, ConcreteCommand
from plumbum.lib import _setdoc, ProcInfo, six
from plumbum.machines.local import LocalPath
from tempfile import NamedTemporaryFile
from plumbum.machines.base import BaseMachine
from plumbum.machines.env import BaseEnv
from plumbum.path.remote import RemotePath, RemoteWorkdir, StatRes


class RemoteEnv(BaseEnv):
    """The remote machine's environment; exposes a dict-like interface"""

    __slots__ = ["_orig", "remote"]

    def __init__(self, remote):
        self.remote = remote
        session = remote._session
        # GNU env has a -0 argument; use it if present. Otherwise,
        # fall back to calling printenv on each (possible) variable
        # from plain env.
        env0 = session.run("env -0; echo")
        if env0[0] == 0 and not env0[2].rstrip():
            self._curr = dict(line.split('=', 1)
                              for line in env0[1].split('\x00')
                              if '=' in line)
        else:
            lines = session.run("env; echo")[1].splitlines()
            split = (line.split('=', 1) for line in lines)
            keys = (line[0] for line in split if len(line)>1)
            runs = ((key, session.run('printenv "%s"; echo' % key))
                    for key in keys)
            self._curr = dict((key, run[1].rstrip('\n')) for (key, run) in runs
                              if run[0] == 0 and run[1].rstrip('\n')
                              and not run[2])
        self._orig = self._curr.copy()
        BaseEnv.__init__(self, self.remote.path, ":")

    @_setdoc(BaseEnv)
    def __delitem__(self, name):
        BaseEnv.__delitem__(self, name)
        self.remote._session.run("unset %s" % (name,))
    @_setdoc(BaseEnv)
    def __setitem__(self, name, value):
        BaseEnv.__setitem__(self, name, value)
        self.remote._session.run("export %s=%s" % (name, shquote(value)))
    @_setdoc(BaseEnv)
    def pop(self, name, *default):
        BaseEnv.pop(self, name, *default)
        self.remote._session.run("unset %s" % (name,))
    @_setdoc(BaseEnv)
    def update(self, *args, **kwargs):
        BaseEnv.update(self, *args, **kwargs)
        self.remote._session.run("export " +
            " ".join("%s=%s" % (k, shquote(v)) for k, v in self.getdict().items()))

    def expand(self, expr):
        """Expands any environment variables and home shortcuts found in ``expr``
        (like ``os.path.expanduser`` combined with ``os.path.expandvars``)

        :param expr: An expression containing environment variables (as ``$FOO``) or
                     home shortcuts (as ``~/.bashrc``)

        :returns: The expanded string"""
        return self.remote.expand(expr)

    def expanduser(self, expr):
        """Expand home shortcuts (e.g., ``~/foo/bar`` or ``~john/foo/bar``)

        :param expr: An expression containing home shortcuts

        :returns: The expanded string"""
        return self.remote.expanduser(expr)

    # def clear(self):
    #    BaseEnv.clear(self, *args, **kwargs)
    #    self.remote._session.run("export %s" % " ".join("%s=%s" % (k, v) for k, v in self.getdict()))

    def getdelta(self):
        """Returns the difference between the this environment and the original environment of
        the remote machine"""
        self._curr["PATH"] = self.path.join()

        delta = {}
        for k, v in self._curr.items():
            if k not in self._orig:
                delta[k] = str(v)
        for k, v in self._orig.items():
            if k not in self._curr:
                delta[k] = ""
            else:
                if v != self._curr[k]:
                    delta[k] = self._curr[k]

        return delta


class RemoteCommand(ConcreteCommand):
    __slots__ = ["remote", "executable"]
    QUOTE_LEVEL = 1

    def __init__(self, remote, executable, encoding = "auto"):
        self.remote = remote
        ConcreteCommand.__init__(self, executable,
            remote.custom_encoding if encoding == "auto" else encoding)
    @property
    def machine(self):
        return self.remote
    def __repr__(self):
        return "RemoteCommand(%r, %r)" % (self.remote, self.executable)
    def popen(self, args = (), **kwargs):
        return self.remote.popen(self[args], **kwargs)
    def nohup(self, cwd='.', stdout='nohup.out', stderr=None, append=True):
        """Runs a command detached."""
        return self.machine.daemonic_popen(self, cwd, stdout, stderr, append)

class ClosedRemoteMachine(Exception):
    pass

class ClosedRemote(object):
    __slots__ = ["_obj", "__weakref__"]
    def __init__(self, obj):
        self._obj = obj
    def close(self):
        pass
    def __getattr__(self, name):
        raise ClosedRemoteMachine("%r has been closed" % (self._obj,))


class BaseRemoteMachine(BaseMachine):
    """Represents a *remote machine*; serves as an entry point to everything related to that
    remote machine, such as working directory and environment manipulation, command creation,
    etc.

    Attributes:

    * ``cwd`` - the remote working directory
    * ``env`` - the remote environment
    * ``custom_encoding`` - the remote machine's default encoding (assumed to be UTF8)
    * ``connect_timeout`` - the connection timeout


    There also is a _cwd attribute that exists if the cwd is not current (del if cwd is changed).
    """

    # allow inheritors to override the RemoteCommand class
    RemoteCommand = RemoteCommand

    @property
    def cwd(self):
        if not hasattr(self, '_cwd'):
            self._cwd = RemoteWorkdir(self)
        return self._cwd

    def __init__(self, encoding = "utf8", connect_timeout = 10, new_session = False):
        self.custom_encoding = encoding
        self.connect_timeout = connect_timeout
        self._session = self.session(new_session = new_session)
        self.uname = self._get_uname()
        self.env = RemoteEnv(self)
        self._python = None

    def _get_uname(self):
        rc, out, _ = self._session.run("uname", retcode = None)
        if rc == 0:
            return out.strip()
        else:
            rc, out, _ = self._session.run("python -c 'import platform;print(platform.uname()[0])'", retcode = None)
            if rc == 0:
                return out.strip()
            else:
                # all POSIX systems should have uname. make an educated guess it's Windows
                return "Windows"

    def __repr__(self):
        return "<%s %s>" % (self.__class__.__name__, self)

    def __enter__(self):
        return self
    def __exit__(self, t, v, tb):
        self.close()
    def close(self):
        """closes the connection to the remote machine; all paths and programs will
        become defunct"""
        self._session.close()
        self._session = ClosedRemote(self)

    def path(self, *parts):
        """A factory for :class:`RemotePaths <plumbum.path.remote.RemotePath>`.
        Usage: ``p = rem.path("/usr", "lib", "python2.7")``
        """
        parts2 = [str(self.cwd)]
        for p in parts:
            if isinstance(p, LocalPath):
                raise TypeError("Cannot construct RemotePath from %r" % (p,))
            parts2.append(self.expanduser(str(p)))
        return RemotePath(self, *parts2)

    def which(self, progname):
        """Looks up a program in the ``PATH``. If the program is not found, raises
        :class:`CommandNotFound <plumbum.commands.CommandNotFound>`

        :param progname: The program's name. Note that if underscores (``_``) are present
                         in the name, and the exact name is not found, they will be replaced
                         in turn by hyphens (``-``) then periods (``.``), and the name will
                         be looked up again for each alternative

        :returns: A :class:`RemotePath <plumbum.path.local.RemotePath>`
        """
        alternatives = [progname]
        if "_" in progname:
            alternatives.append(progname.replace("_", "-"))
            alternatives.append(progname.replace("_", "."))
        for name in alternatives:
            for p in self.env.path:
                fn = p / name
                if fn.access("x") and not fn.is_dir():
                    return fn

        raise CommandNotFound(progname, self.env.path)

    def __getitem__(self, cmd):
        """Returns a `Command` object representing the given program. ``cmd`` can be a string or
        a :class:`RemotePath <plumbum.path.remote.RemotePath>`; if it is a path, a command
        representing this path will be returned; otherwise, the program name will be looked up in
        the system's ``PATH`` (using ``which``). Usage::

            r_ls = rem["ls"]
        """
        if isinstance(cmd, RemotePath):
            if cmd.remote is self:
                return self.RemoteCommand(self, cmd)
            else:
                raise TypeError("Given path does not belong to this remote machine: %r" % (cmd,))
        elif not isinstance(cmd, LocalPath):
            if "/" in cmd or "\\" in cmd:
                return self.RemoteCommand(self, self.path(cmd))
            else:
                return self.RemoteCommand(self, self.which(cmd))
        else:
            raise TypeError("cmd must not be a LocalPath: %r" % (cmd,))

    @property
    def python(self):
        """A command that represents the default remote python interpreter"""
        if not self._python:
            self._python = self["python"]
        return self._python

    def session(self, isatty = False, new_session = False):
        """Creates a new :class:`ShellSession <plumbum.session.ShellSession>` object; this invokes the user's
        shell on the remote machine and executes commands on it over stdin/stdout/stderr"""
        raise NotImplementedError()

    def download(self, src, dst):
        """Downloads a remote file/directory (``src``) to a local destination (``dst``).
        ``src`` must be a string or a :class:`RemotePath <plumbum.path.remote.RemotePath>`
        pointing to this remote machine, and ``dst`` must be a string or a
        :class:`LocalPath <plumbum.machines.local.LocalPath>`"""
        raise NotImplementedError()

    def upload(self, src, dst):
        """Uploads a local file/directory (``src``) to a remote destination (``dst``).
        ``src`` must be a string or a :class:`LocalPath <plumbum.machines.local.LocalPath>`,
        and ``dst`` must be a string or a :class:`RemotePath <plumbum.path.remote.RemotePath>`
        pointing to this remote machine"""
        raise NotImplementedError()

    def popen(self, args, **kwargs):
        """Spawns the given command on the remote machine, returning a ``Popen``-like object;
        do not use this method directly, unless you need "low-level" control on the remote
        process"""
        raise NotImplementedError()

    def list_processes(self):
        """
        Returns information about all running processes (on POSIX systems: using ``ps``)

        .. versionadded:: 1.3
        """
        ps = self["ps"]
        lines = ps("-e", "-o", "pid,uid,stat,args").splitlines()
        lines.pop(0) # header
        for line in lines:
            parts = line.strip().split()
            yield ProcInfo(int(parts[0]), int(parts[1]), parts[2], " ".join(parts[3:]))

    def pgrep(self, pattern):
        """
        Process grep: return information about all processes whose command-line args match the given regex pattern
        """
        pat = re.compile(pattern)
        for procinfo in self.list_processes():
            if pat.search(procinfo.args):
                yield procinfo

    @contextmanager
    def tempdir(self):
        """A context manager that creates a remote temporary directory, which is removed when
        the context exits"""
        _, out, _ = self._session.run("mktemp -d tmp.XXXXXXXXXX")
        dir = self.path(out.strip())  # @ReservedAssignment
        try:
            yield dir
        finally:
            dir.delete()

    #
    # Path implementation
    #
    def _path_listdir(self, fn):
        files = self._session.run("ls -a %s" % (shquote(fn),))[1].splitlines()
        files.remove(".")
        files.remove("..")
        return files
    def _path_glob(self, fn, pattern):
        # shquote does not work here due to the way bash loops use space as a seperator
        pattern = pattern.replace(" ", r"\ ")
        fn = fn.replace(" ", r"\ ")
        matches = self._session.run(r'for fn in {0}/{1}; do echo $fn; done'.format(fn,pattern))[1].splitlines()
        if len(matches) == 1 and not self._path_stat(matches[0]):
            return []  # pattern expansion failed
        return matches

    def _path_getuid(self, fn):
        stat_cmd = "stat -c '%u,%U' " if self.uname not in ('Darwin', 'FreeBSD') else "stat -f '%u,%Su' "
        return self._session.run(stat_cmd + shquote(fn))[1].strip().split(",")
    def _path_getgid(self, fn):
        stat_cmd = "stat -c '%g,%G' " if self.uname not in ('Darwin', 'FreeBSD') else "stat -f '%g,%Sg' "
        return self._session.run(stat_cmd + shquote(fn))[1].strip().split(",")
    def _path_stat(self, fn):
        if self.uname not in ('Darwin', 'FreeBSD'):
            stat_cmd = "stat -c '%F,%f,%i,%d,%h,%u,%g,%s,%X,%Y,%Z' "
        else:
            stat_cmd = "stat -f '%HT,%Xp,%i,%d,%l,%u,%g,%z,%a,%m,%c' "
        rc, out, _ = self._session.run(stat_cmd + shquote(fn), retcode = None)
        if rc != 0:
            return None
        statres = out.strip().split(",")
        text_mode = statres.pop(0).lower()
        res = StatRes((int(statres[0], 16),) + tuple(int(sr) for sr in statres[1:]))
        res.text_mode = text_mode
        return res

    def _path_delete(self, fn):
        self._session.run("rm -rf %s" % (shquote(fn),))
    def _path_move(self, src, dst):
        self._session.run("mv %s %s" % (shquote(src), shquote(dst)))
    def _path_copy(self, src, dst):
        self._session.run("cp -r %s %s" % (shquote(src), shquote(dst)))
    def _path_mkdir(self, fn):
        self._session.run("mkdir -p %s" % (shquote(fn),))
    def _path_chmod(self, mode, fn):
        self._session.run("chmod %o %s" % (mode, shquote(fn)))
    def _path_touch(self, path):
        self._session.run("touch {path}".format(path=path))
    def _path_chown(self, fn, owner, group, recursive):
        args = ["chown"]
        if recursive:
            args.append("-R")
        if owner is not None and group is not None:
            args.append("%s:%s" % (owner, group))
        elif owner is not None:
            args.append(str(owner))
        elif group is not None:
            args.append(":%s" % (group,))
        args.append(shquote(fn))
        self._session.run(" ".join(args))

    def _path_read(self, fn):
        data = self["cat"](fn)
        if self.custom_encoding and isinstance(data, six.unicode_type):
            data = data.encode(self.custom_encoding)
        return data
    def _path_write(self, fn, data):
        if self.custom_encoding and isinstance(data, six.unicode_type):
            data = data.encode(self.custom_encoding)
        with NamedTemporaryFile() as f:
            f.write(data)
            f.flush()
            f.seek(0)
            self.upload(f.name, fn)

    def _path_link(self, src, dst, symlink):
        self._session.run("ln %s %s %s" % ("-s" if symlink else "", shquote(src), shquote(dst)))

    @_setdoc(BaseEnv)
    def expand(self, expr):
        return self._session.run("echo %s" % (expr,))[1].strip()

    @_setdoc(BaseEnv)
    def expanduser(self, expr):
        if not any(part.startswith("~") for part in expr.split("/")):
            return expr
        # we escape all $ signs to avoid expanding env-vars
        return self._session.run("echo %s" % (expr.replace("$", "\\$"),))[1].strip()

1	import re	8✔
2	from contextlib import contextmanager	8✔
3	from plumbum.commands import CommandNotFound, shquote, ConcreteCommand	8✔
4	from plumbum.lib import _setdoc, ProcInfo, six	8✔
5	from plumbum.machines.local import LocalPath	8✔
6	from tempfile import NamedTemporaryFile	8✔
7	from plumbum.machines.base import BaseMachine	8✔
8	from plumbum.machines.env import BaseEnv	8✔
9	from plumbum.path.remote import RemotePath, RemoteWorkdir, StatRes	8✔
10
11
12	class RemoteEnv(BaseEnv):	8✔
13	"""The remote machine's environment; exposes a dict-like interface"""
14
15	__slots__ = ["_orig", "remote"]	8✔
16
17	def __init__(self, remote):	8✔
18	self.remote = remote	8✔
19	session = remote._session	8✔
20	# GNU env has a -0 argument; use it if present. Otherwise,
21	# fall back to calling printenv on each (possible) variable
22	# from plain env.
23	env0 = session.run("env -0; echo")	8✔
24	if env0[0] == 0 and not env0[2].rstrip():	8✔
25	self._curr = dict(line.split('=', 1)	6 all except PYV=Mac2 and PY3=3 PYV=Mac3 ✔
26	for line in env0[1].split('\x00')
27	if '=' in line)
28	else:
29	lines = session.run("env; echo")[1].splitlines()	2 only PYV=Mac2 and PY3=3 PYV=Mac3 ✔
30	split = (line.split('=', 1) for line in lines)	2 only PYV=Mac2 and PY3=3 PYV=Mac3 ✔
31	keys = (line[0] for line in split if len(line)>1)	2 only PYV=Mac2 and PY3=3 PYV=Mac3 ✔
32	runs = ((key, session.run('printenv "%s"; echo' % key))	2 only PYV=Mac2 and PY3=3 PYV=Mac3 ✔
33	for key in keys)
34	self._curr = dict((key, run[1].rstrip('\n')) for (key, run) in runs	2 only PYV=Mac2 and PY3=3 PYV=Mac3 ✔
35	if run[0] == 0 and run[1].rstrip('\n')
36	and not run[2])
37	self._orig = self._curr.copy()	8✔
38	BaseEnv.__init__(self, self.remote.path, ":")	8✔
39
40	@_setdoc(BaseEnv)	8✔
41	def __delitem__(self, name):
42	BaseEnv.__delitem__(self, name)	×
43	self.remote._session.run("unset %s" % (name,))	×
44	@_setdoc(BaseEnv)	8✔
45	def __setitem__(self, name, value):
46	BaseEnv.__setitem__(self, name, value)	×
47	self.remote._session.run("export %s=%s" % (name, shquote(value)))	×
48	@_setdoc(BaseEnv)	8✔
49	def pop(self, name, *default):
50	BaseEnv.pop(self, name, *default)	×
51	self.remote._session.run("unset %s" % (name,))	×
52	@_setdoc(BaseEnv)	8✔
53	def update(self, args, *kwargs):
54	BaseEnv.update(self, args, *kwargs)	8✔
55	self.remote._session.run("export " +	8✔
56	" ".join("%s=%s" % (k, shquote(v)) for k, v in self.getdict().items()))
57
58	def expand(self, expr):	8✔
59	"""Expands any environment variables and home shortcuts found in ``expr``
60	(like ``os.path.expanduser`` combined with ``os.path.expandvars``)
61
62	:param expr: An expression containing environment variables (as ``$FOO``) or
63	home shortcuts (as ``~/.bashrc``)
64
65	:returns: The expanded string"""
66	return self.remote.expand(expr)	×
67
68	def expanduser(self, expr):	8✔
69	"""Expand home shortcuts (e.g., ``~/foo/bar`` or ``~john/foo/bar``)
70
71	:param expr: An expression containing home shortcuts
72
73	:returns: The expanded string"""
74	return self.remote.expanduser(expr)	×
75
76	# def clear(self):
77	# BaseEnv.clear(self, args, *kwargs)
78	# self.remote._session.run("export %s" % " ".join("%s=%s" % (k, v) for k, v in self.getdict()))
79
80	def getdelta(self):	8✔
81	"""Returns the difference between the this environment and the original environment of
82	the remote machine"""
83	self._curr["PATH"] = self.path.join()	8✔
84
85	delta = {}	8✔
86	for k, v in self._curr.items():	8✔
87	if k not in self._orig:	8✔
88	delta[k] = str(v)	8✔
89	for k, v in self._orig.items():	8✔
90	if k not in self._curr:	8!
91	delta[k] = ""	×
92	else:
93	if v != self._curr[k]:	8!
94	delta[k] = self._curr[k]	×
95
96	return delta	8✔
97
98
99	class RemoteCommand(ConcreteCommand):	8✔
100	__slots__ = ["remote", "executable"]	8✔
101	QUOTE_LEVEL = 1	8✔
102
103	def __init__(self, remote, executable, encoding = "auto"):	8✔
104	self.remote = remote	8✔
105	ConcreteCommand.__init__(self, executable,	8✔
106	remote.custom_encoding if encoding == "auto" else encoding)
107	@property	8✔
108	def machine(self):
109	return self.remote	8✔
110	def __repr__(self):
111	return "RemoteCommand(%r, %r)" % (self.remote, self.executable)
112	def popen(self, args = (), **kwargs):	8✔
113	return self.remote.popen(self[args], **kwargs)	8✔
114	def nohup(self, cwd='.', stdout='nohup.out', stderr=None, append=True):	8✔
115	"""Runs a command detached."""
116	return self.machine.daemonic_popen(self, cwd, stdout, stderr, append)	×
117
118	class ClosedRemoteMachine(Exception):	8✔
119	pass	8✔
120
121	class ClosedRemote(object):	8✔
122	__slots__ = ["_obj", "__weakref__"]	8✔
123	def __init__(self, obj):	8✔
124	self._obj = obj	8✔
125	def close(self):	8✔
126	pass	×
127	def __getattr__(self, name):	8✔
128	raise ClosedRemoteMachine("%r has been closed" % (self._obj,))	×
129
130
131	class BaseRemoteMachine(BaseMachine):	8✔
132	"""Represents a remote machine; serves as an entry point to everything related to that
133	remote machine, such as working directory and environment manipulation, command creation,
134	etc.
135
136	Attributes:
137
138	* ``cwd`` - the remote working directory
139	* ``env`` - the remote environment
140	* ``custom_encoding`` - the remote machine's default encoding (assumed to be UTF8)
141	* ``connect_timeout`` - the connection timeout
142
143
144	There also is a _cwd attribute that exists if the cwd is not current (del if cwd is changed).
145	"""
146
147	# allow inheritors to override the RemoteCommand class
148	RemoteCommand = RemoteCommand	8✔
149
150	@property	8✔
151	def cwd(self):
152	if not hasattr(self, '_cwd'):	8✔
153	self._cwd = RemoteWorkdir(self)	8✔
154	return self._cwd	8✔
155
156	def __init__(self, encoding = "utf8", connect_timeout = 10, new_session = False):	8✔
157	self.custom_encoding = encoding	8✔
158	self.connect_timeout = connect_timeout	8✔
159	self._session = self.session(new_session = new_session)	8✔
160	self.uname = self._get_uname()	8✔
161	self.env = RemoteEnv(self)	8✔
162	self._python = None	8✔
163
164	def _get_uname(self):	8✔
165	rc, out, _ = self._session.run("uname", retcode = None)	8✔
166	if rc == 0:	8!
167	return out.strip()	8✔
168	else:
169	rc, out, _ = self._session.run("python -c 'import platform;print(platform.uname()[0])'", retcode = None)	×
170	if rc == 0:	×
171	return out.strip()	×
172	else:
173	# all POSIX systems should have uname. make an educated guess it's Windows
174	return "Windows"	×
175
176	def __repr__(self):
177	return "<%s %s>" % (self.__class__.__name__, self)
178
179	def __enter__(self):	8✔
180	return self	8✔
181	def __exit__(self, t, v, tb):	8✔
182	self.close()	8✔
183	def close(self):	8✔
184	"""closes the connection to the remote machine; all paths and programs will
185	become defunct"""
186	self._session.close()	8✔
187	self._session = ClosedRemote(self)	8✔
188
189	def path(self, *parts):	8✔
190	"""A factory for :class:`RemotePaths <plumbum.path.remote.RemotePath>`.
191	Usage: ``p = rem.path("/usr", "lib", "python2.7")``
192	"""
193	parts2 = [str(self.cwd)]	8✔
194	for p in parts:	8✔
195	if isinstance(p, LocalPath):	8!
196	raise TypeError("Cannot construct RemotePath from %r" % (p,))	×
197	parts2.append(self.expanduser(str(p)))	8✔
198	return RemotePath(self, *parts2)	8✔
199
200	def which(self, progname):	8✔
201	"""Looks up a program in the ``PATH``. If the program is not found, raises
202	:class:`CommandNotFound <plumbum.commands.CommandNotFound>`
203
204	:param progname: The program's name. Note that if underscores (``_``) are present
205	in the name, and the exact name is not found, they will be replaced
206	in turn by hyphens (``-``) then periods (``.``), and the name will
207	be looked up again for each alternative
208
209	:returns: A :class:`RemotePath <plumbum.path.local.RemotePath>`
210	"""
211	alternatives = [progname]	8✔
212	if "_" in progname:	8✔
213	alternatives.append(progname.replace("_", "-"))	8✔
214	alternatives.append(progname.replace("_", "."))	8✔
215	for name in alternatives:	8✔
216	for p in self.env.path:	8✔
217	fn = p / name	8✔
218	if fn.access("x") and not fn.is_dir():	8✔
219	return fn	8✔
220
221	raise CommandNotFound(progname, self.env.path)	8✔
222
223	def __getitem__(self, cmd):	8✔
224	"""Returns a `Command` object representing the given program. ``cmd`` can be a string or
225	a :class:`RemotePath <plumbum.path.remote.RemotePath>`; if it is a path, a command
226	representing this path will be returned; otherwise, the program name will be looked up in
227	the system's ``PATH`` (using ``which``). Usage::
228
229	r_ls = rem["ls"]
230	"""
231	if isinstance(cmd, RemotePath):	8!
232	if cmd.remote is self:	×
233	return self.RemoteCommand(self, cmd)	×
234	else:
235	raise TypeError("Given path does not belong to this remote machine: %r" % (cmd,))	×
236	elif not isinstance(cmd, LocalPath):	8!
237	if "/" in cmd or "\\" in cmd:	8!
238	return self.RemoteCommand(self, self.path(cmd))	×
239	else:
240	return self.RemoteCommand(self, self.which(cmd))	8✔
241	else:
242	raise TypeError("cmd must not be a LocalPath: %r" % (cmd,))	×
243
244	@property	8✔
245	def python(self):
246	"""A command that represents the default remote python interpreter"""
247	if not self._python:	8✔
248	self._python = self["python"]	8✔
249	return self._python	8✔
250
251	def session(self, isatty = False, new_session = False):	8✔
252	"""Creates a new :class:`ShellSession <plumbum.session.ShellSession>` object; this invokes the user's
253	shell on the remote machine and executes commands on it over stdin/stdout/stderr"""
254	raise NotImplementedError()
255
256	def download(self, src, dst):	8✔
257	"""Downloads a remote file/directory (``src``) to a local destination (``dst``).
258	``src`` must be a string or a :class:`RemotePath <plumbum.path.remote.RemotePath>`
259	pointing to this remote machine, and ``dst`` must be a string or a
260	:class:`LocalPath <plumbum.machines.local.LocalPath>`"""
261	raise NotImplementedError()
262
263	def upload(self, src, dst):	8✔
264	"""Uploads a local file/directory (``src``) to a remote destination (``dst``).
265	``src`` must be a string or a :class:`LocalPath <plumbum.machines.local.LocalPath>`,
266	and ``dst`` must be a string or a :class:`RemotePath <plumbum.path.remote.RemotePath>`
267	pointing to this remote machine"""
268	raise NotImplementedError()
269
270	def popen(self, args, **kwargs):	8✔
271	"""Spawns the given command on the remote machine, returning a ``Popen``-like object;
272	do not use this method directly, unless you need "low-level" control on the remote
273	process"""
274	raise NotImplementedError()
275
276	def list_processes(self):	8✔
277	"""
278	Returns information about all running processes (on POSIX systems: using ``ps``)
279
280	.. versionadded:: 1.3
281	"""
282	ps = self["ps"]	8✔
283	lines = ps("-e", "-o", "pid,uid,stat,args").splitlines()	8✔
284	lines.pop(0) # header	8✔
285	for line in lines:	8✔
286	parts = line.strip().split()	8✔
287	yield ProcInfo(int(parts[0]), int(parts[1]), parts[2], " ".join(parts[3:]))	8✔
288
289	def pgrep(self, pattern):	8✔
290	"""
291	Process grep: return information about all processes whose command-line args match the given regex pattern
292	"""
293	pat = re.compile(pattern)	8✔
294	for procinfo in self.list_processes():	8✔
295	if pat.search(procinfo.args):	8✔
296	yield procinfo	8✔
297
298	@contextmanager	8✔
299	def tempdir(self):
300	"""A context manager that creates a remote temporary directory, which is removed when
301	the context exits"""
302	_, out, _ = self._session.run("mktemp -d tmp.XXXXXXXXXX")	8✔
303	dir = self.path(out.strip()) # @ReservedAssignment	8✔
304	try:	8✔
305	yield dir	8✔
306	finally:
307	dir.delete()	8✔
308
309	#
310	# Path implementation
311	#
312	def _path_listdir(self, fn):	8✔
313	files = self._session.run("ls -a %s" % (shquote(fn),))[1].splitlines()	8✔
314	files.remove(".")	8✔
315	files.remove("..")	8✔
316	return files	8✔
317	def _path_glob(self, fn, pattern):	8✔
318	# shquote does not work here due to the way bash loops use space as a seperator
319	pattern = pattern.replace(" ", r"\ ")	8✔
320	fn = fn.replace(" ", r"\ ")	8✔
321	matches = self._session.run(r'for fn in {0}/{1}; do echo $fn; done'.format(fn,pattern))[1].splitlines()	8✔
322	if len(matches) == 1 and not self._path_stat(matches[0]):	8!
323	return [] # pattern expansion failed	×
324	return matches	8✔
325
326	def _path_getuid(self, fn):	8✔
327	stat_cmd = "stat -c '%u,%U' " if self.uname not in ('Darwin', 'FreeBSD') else "stat -f '%u,%Su' "	8✔
328	return self._session.run(stat_cmd + shquote(fn))[1].strip().split(",")	8✔
329	def _path_getgid(self, fn):	8✔
330	stat_cmd = "stat -c '%g,%G' " if self.uname not in ('Darwin', 'FreeBSD') else "stat -f '%g,%Sg' "	8✔
331	return self._session.run(stat_cmd + shquote(fn))[1].strip().split(",")	8✔
332	def _path_stat(self, fn):	8✔
333	if self.uname not in ('Darwin', 'FreeBSD'):	8✔
334	stat_cmd = "stat -c '%F,%f,%i,%d,%h,%u,%g,%s,%X,%Y,%Z' "	6 all except PYV=Mac2 and PY3=3 PYV=Mac3 ✔
335	else:
336	stat_cmd = "stat -f '%HT,%Xp,%i,%d,%l,%u,%g,%z,%a,%m,%c' "	2 only PYV=Mac2 and PY3=3 PYV=Mac3 ✔
337	rc, out, _ = self._session.run(stat_cmd + shquote(fn), retcode = None)	8✔
338	if rc != 0:	8✔
339	return None	8✔
340	statres = out.strip().split(",")	8✔
341	text_mode = statres.pop(0).lower()	8✔
342	res = StatRes((int(statres[0], 16),) + tuple(int(sr) for sr in statres[1:]))	8✔
343	res.text_mode = text_mode	8✔
344	return res	8✔
345
346	def _path_delete(self, fn):	8✔
347	self._session.run("rm -rf %s" % (shquote(fn),))	8✔
348	def _path_move(self, src, dst):	8✔
349	self._session.run("mv %s %s" % (shquote(src), shquote(dst)))	×
350	def _path_copy(self, src, dst):	8✔
351	self._session.run("cp -r %s %s" % (shquote(src), shquote(dst)))	8✔
352	def _path_mkdir(self, fn):	8✔
353	self._session.run("mkdir -p %s" % (shquote(fn),))	×
354	def _path_chmod(self, mode, fn):	8✔
355	self._session.run("chmod %o %s" % (mode, shquote(fn)))	×
356	def _path_touch(self, path):	8✔
357	self._session.run("touch {path}".format(path=path))	8✔
358	def _path_chown(self, fn, owner, group, recursive):	8✔
359	args = ["chown"]	8✔
360	if recursive:	8!
361	args.append("-R")	×
362	if owner is not None and group is not None:	8!
363	args.append("%s:%s" % (owner, group))	×
364	elif owner is not None:	8!
365	args.append(str(owner))	8✔
366	elif group is not None:	×
367	args.append(":%s" % (group,))	×
368	args.append(shquote(fn))	8✔
369	self._session.run(" ".join(args))	8✔
370
371	def _path_read(self, fn):	8✔
372	data = self["cat"](fn)	8✔
373	if self.custom_encoding and isinstance(data, six.unicode_type):	8!
374	data = data.encode(self.custom_encoding)	8✔
375	return data	8✔
376	def _path_write(self, fn, data):	8✔
377	if self.custom_encoding and isinstance(data, six.unicode_type):	8!
378	data = data.encode(self.custom_encoding)	×
379	with NamedTemporaryFile() as f:	8✔
380	f.write(data)	8✔
381	f.flush()	8✔
382	f.seek(0)	8✔
383	self.upload(f.name, fn)	8✔
384
385	def _path_link(self, src, dst, symlink):	8✔
386	self._session.run("ln %s %s %s" % ("-s" if symlink else "", shquote(src), shquote(dst)))	×
387
388	@_setdoc(BaseEnv)	8✔
389	def expand(self, expr):
390	return self._session.run("echo %s" % (expr,))[1].strip()	×
391
392	@_setdoc(BaseEnv)	8✔
393	def expanduser(self, expr):
394	if not any(part.startswith("~") for part in expr.split("/")):	8!
395	return expr	8✔
396	# we escape all $ signs to avoid expanding env-vars
397	return self._session.run("echo %s" % (expr.replace("$", "\\$"),))[1].strip()	×

tomerfiliba / plumbum / 872

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous