diff --git a/plumbum/cmd.py b/plumbum/cmd.py new file mode 100644 index 000000000..875dec7e0 --- /dev/null +++ b/plumbum/cmd.py @@ -0,0 +1,20 @@ +""" +Module hack: ``from plumbum.cmd import ls`` +""" +import sys +from types import ModuleType +from plumbum.local_machine import local + +__all__ = [] + +class LocalModule(ModuleType): + """The module-hack that allows us to use ``from plumbum.cmd import some_program``""" + def __init__(self, name): + ModuleType.__init__(self, name, __doc__) + self.__file__ = None + self.__package__ = ".".join(name.split(".")[:-1]) + def __getattr__(self, name): + return local[name] + +LocalModule = LocalModule("plumbum.cmd") +sys.modules[LocalModule.__name__] = LocalModule diff --git a/plumbum/lib.py b/plumbum/lib.py index 04575ebab..7aa406165 100644 --- a/plumbum/lib.py +++ b/plumbum/lib.py @@ -1,7 +1,7 @@ import heapq -def _setdoc(super): +def _setdoc(super): #@ReservedAssignment def deco(func): func.__doc__ = getattr(getattr(super, func.__name__, None), "__doc__", None) return func diff --git a/plumbum/local_machine.py b/plumbum/local_machine.py index 80be3f163..a6269c063 100644 --- a/plumbum/local_machine.py +++ b/plumbum/local_machine.py @@ -1,24 +1,31 @@ from __future__ import with_statement import os import sys -import grp -import pwd import glob import shutil import subprocess import logging import stat import time -from types import ModuleType from tempfile import mkdtemp from subprocess import Popen, PIPE from contextlib import contextmanager -from plumbum.path import Path +from plumbum.path import Path, FSUser from plumbum.remote_path import RemotePath from plumbum.commands import CommandNotFound, ConcreteCommand from plumbum.session import ShellSession from plumbum.lib import _setdoc +import platform + +try: + from pwd import getpwuid, getpwnam + from grp import getgrgid, getgrnam +except ImportError: + def getpwuid(x): return (None,) + def getgrgid(x): return (None,) + def getpwnam(x): raise OSError("`getpwnam` not supported") + def getgrnam(x): raise OSError("`getgrnam` not supported") logger = logging.getLogger("plumbum.local") @@ -60,23 +67,17 @@ def dirname(self): @property @_setdoc(Path) - def owner(self): - stat = os.stat(str(self)) - return pwd.getpwuid(stat.st_uid)[0] - - @owner.setter - def owner(self, owner): - self.chown(owner) + def uid(self): + uid = self.stat().st_uid + name = getpwuid(uid)[0] + return FSUser(uid, name) @property @_setdoc(Path) def group(self): - stat = os.stat(str(self)) - return grp.getgrgid(stat.st_gid)[0] - - @group.setter - def group(self, group): - self.chown(group=group) + gid = self.stat().st_gid + name = getgrgid(gid)[0] + return FSUser(gid, name) @_setdoc(Path) def join(self, other): @@ -157,22 +158,15 @@ def write(self, data): f.write(data) @_setdoc(Path) - def chown(self, owner='', group='', uid='', gid='', recursive=False): - gid = str(gid) # str so uid 0 (int) isn't seen as False - uid = str(uid) - args = list() - if recursive: - args.append('-R') - if uid: - owner = uid - if gid: - group = gid - if group: - owner = '%s:%s' % (owner, group) - args.append(owner) - args.append(str(self)) - # recursive is a pain using os.chown - local['chown'](*args) + def chown(self, owner=None, group=None, recursive = None): + if not hasattr(os, "chown"): + raise OSError("os.chown() not supported") + uid = owner if isinstance(owner, int) else getpwnam(owner)[2] + gid = group if isinstance(group, int) else getgrnam(group)[2] + os.chown(str(self), uid, gid) + if recursive or (recursive is None and self.isdir()): + for subpath in self.walk(): + os.chown(str(subpath), uid, gid) class Workdir(LocalPath): @@ -449,6 +443,7 @@ class LocalMachine(object): cwd = Workdir() env = LocalEnv() encoding = sys.getfilesystemencoding() + uname = platform.uname()[0] if IS_WIN32: _EXTENSIONS = [""] + env.get("PATHEXT", ":.exe:.bat").lower().split(os.path.pathsep) @@ -502,10 +497,9 @@ def which(cls, progname): raise CommandNotFound(progname, list(cls.env.path)) def path(self, *parts): - """A factory for :class:`LocalPaths `. Usage - - :: - + """A factory for :class:`LocalPaths `. + Usage :: + p = local.path("/usr", "lib", "python2.7") """ parts2 = [str(self.cwd)] @@ -565,7 +559,7 @@ def session(self): def tempdir(self): """A context manager that creates a temporary directory, which is removed when the context exits""" - dir = self.path(mkdtemp()) + dir = self.path(mkdtemp()) #@ReservedAssignment try: yield dir finally: @@ -586,18 +580,3 @@ def tempdir(self): * ``encoding`` - the local machine's default encoding (``sys.getfilesystemencoding()``) """ -#=================================================================================================== -# Module hack: ``from plumbum.cmd import ls`` -#=================================================================================================== -class LocalModule(ModuleType): - """The module-hack that allows us to use ``from plumbum.cmd import some_program``""" - def __init__(self, name): - ModuleType.__init__(self, name, __doc__) - self.__file__ = None - self.__package__ = ".".join(name.split(".")[:-1]) - def __getattr__(self, name): - return local[name] - -LocalModule = LocalModule("plumbum.cmd") -sys.modules[LocalModule.__name__] = LocalModule - diff --git a/plumbum/path.py b/plumbum/path.py index 44652b699..00fad80ec 100644 --- a/plumbum/path.py +++ b/plumbum/path.py @@ -1,3 +1,14 @@ +class FSUser(int): + """A special object that represents file-system user. It derives from ``int``, so it behaves + just like a number (``uid``/``gid``), but also have a ``.name`` attribute that holds the + string-name of the user, if given + """ + __slots__ = ["name"] + def __new__(cls, val, name = None): + self = int.__new__(cls, val) + self.name = name + return self + class Path(object): """An abstraction over file system paths. This class is abstract, and the two implementations are :class:`LocalPath ` and @@ -68,12 +79,16 @@ def dirname(self): """The dirname component of this path""" raise NotImplementedError() @property - def owner(self): - """The owner of leaf component of this path""" + def uid(self): + """The user that owns this path. The returned value is a :class:`FSUser ` + object which behaves like an ``int`` (as expected from ``uid``), but it also has a ``.name`` + attribute that holds the string-name of the user""" raise NotImplementedError() @property - def group(self): - """The group of leaf component of this path""" + def gid(self): + """The group that owns this path. The returned value is a :class:`FSUser ` + object which behaves like an ``int`` (as expected from ``gid``), but it also has a ``.name`` + attribute that holds the string-name of the group""" raise NotImplementedError() def _get_info(self): @@ -122,6 +137,14 @@ def read(self): def write(self, data): """writes the given data to this file""" raise NotImplementedError() - def chown(self, owner=None, group=None, uid=None, gid=None, recursive=False): - """Change ownership of leaf component of this path""" + def chown(self, owner=None, group=None, recursive=None): + """Change ownership of this path. + + :param owner: The owner to set (either ``uid`` or ``username``), optional + :param owner: The group to set (either ``gid`` or ``groupname``), optional + :param recursive: whether to change ownership of all contained files and subdirectories. + Only meaningful when ``self`` is a directory. If ``None``, the value + will default to ``True`` if ``self`` is a directory, ``False`` otherwise. + """ raise NotImplementedError() + diff --git a/plumbum/remote_machine.py b/plumbum/remote_machine.py index c44445489..35d102d32 100644 --- a/plumbum/remote_machine.py +++ b/plumbum/remote_machine.py @@ -180,10 +180,9 @@ def close(self): self._session = ClosedRemote(self) def path(self, *parts): - """A factory for :class:`RemotePaths `. Usage - - :: - + """A factory for :class:`RemotePaths `. + Usage :: + p = rem.path("/usr", "lib", "python2.7") """ parts2 = [str(self.cwd)] @@ -274,7 +273,7 @@ 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") - dir = self.path(out.strip()) + dir = self.path(out.strip()) #@ReservedAssignment try: yield dir finally: diff --git a/plumbum/remote_path.py b/plumbum/remote_path.py index fe20e95de..611f80d95 100644 --- a/plumbum/remote_path.py +++ b/plumbum/remote_path.py @@ -3,7 +3,7 @@ import errno import six from tempfile import NamedTemporaryFile -from plumbum.path import Path +from plumbum.path import Path, FSUser from plumbum.lib import _setdoc from plumbum.commands import shquote @@ -61,24 +61,15 @@ def dirname(self): @property @_setdoc(Path) - def owner(self): - files = self.remote._session.run("ls -a %s" % (self,))[1].splitlines() - stat = os.stat(str(self)) - return pwd.getpwuid(stat.st_uid)[0] - - @owner.setter - def owner(self, owner): - self.chown(owner) + def uid(self): + uid, name = self.remote._session.run("stat -c '%u,%U' " + shquote(self)).split(",") + return FSUser(int(uid), name) @property @_setdoc(Path) - def group(self): - stat = os.stat(str(self)) - return grp.getgrgid(stat.st_gid)[0] - - @group.setter - def group(self, group): - self.chown(group=group) + def gid(self): + gid, name = self.remote._session.run("stat -c '%g,%G' " + shquote(self)).split(",") + return FSUser(int(gid), name) def _get_info(self): return (self.remote, self._path) @@ -183,17 +174,18 @@ def write(self, data): self.remote.upload(f.name, self) @_setdoc(Path) - def chown(self, owner='', group='', uid='', gid='', recursive=False): - gid = str(gid) # str so uid 0 (int) isn't seen as False - uid = str(uid) - args = list() + def chown(self, owner=None, group=None, recursive=None): + args = ["chown"] + if recursive is None: + recursive = self.isdir() if recursive: - args.append('-R') - if uid: - owner = uid - if gid: - group = gid - if group: - owner = '%s:%s' % (owner, group) - args.append(owner) - self.remote._session.run('chown %s "%s"' % (' '.join(args), self)) + 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(self)) + self.remote._session.run(" ".join(args)) + diff --git a/plumbum/version.py b/plumbum/version.py index d4f021536..14884a4f3 100644 --- a/plumbum/version.py +++ b/plumbum/version.py @@ -1,3 +1,3 @@ version = (1, 0, 0) version_string = "1.0.0" -release_date = "2012.08.01" +release_date = "2012.10.06" diff --git a/tests/test_local.py b/tests/test_local.py index d5b945080..dab6b6714 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -1,215 +1,215 @@ -from __future__ import with_statement -import os -import unittest -import six -from plumbum import local, LocalPath, FG, BG, ERROUT -from plumbum import CommandNotFound, ProcessExecutionError, ProcessTimedOut - - -class LocalPathTest(unittest.TestCase): - def test_basename(self): - name = LocalPath("/some/long/path/to/file.txt").basename - self.assertTrue(isinstance(name, six.string_types)) - self.assertEqual("file.txt", str(name)) - - def test_dirname(self): - name = LocalPath("/some/long/path/to/file.txt").dirname - self.assertTrue(isinstance(name, LocalPath)) - self.assertEqual("/some/long/path/to", str(name)) - - # requires being run as root - def test_chown(self): - path = LocalPath("/tmp/delme.txt") - path.delete() - path.write('test') - self.assertTrue('nobody' != path.owner) - self.assertTrue('nogroup' != path.group) - # chown group - path.chown(group='nogroup') - self.assertEqual('nogroup', path.group) - self.assertTrue('nobody' != path.owner) - # chown owner - path.chown('nobody') - self.assertEqual('nobody', path.owner) - # chown both / numerical ids - path.chown(uid=0, gid=0) - self.assertEqual('root', path.owner) - self.assertEqual('root', path.group) - # recursive - path.chown('root', recursive=True) - # set properties - path.owner = 'nobody' - self.assertEqual('nobody', path.owner) - path.group = 'nogroup' - self.assertEqual('nogroup', path.group) - path.delete() - - -class LocalMachineTest(unittest.TestCase): - def test_imports(self): - from plumbum.cmd import ls - self.assertTrue("test_local.py" in local["ls"]().splitlines()) - self.assertTrue("test_local.py" in ls().splitlines()) - - self.assertRaises(CommandNotFound, lambda: local["non_exist1N9"]) - - try: - from plumbum.cmd import non_exist1N9 #@UnresolvedImport @UnusedImport - except CommandNotFound: - pass - else: - self.fail("from plumbum.cmd import non_exist1N9") - - def test_cwd(self): - from plumbum.cmd import ls - self.assertEqual(local.cwd, os.getcwd()) - self.assertTrue("__init__.py" not in ls().splitlines()) - with local.cwd("../plumbum"): - self.assertTrue("__init__.py" in ls().splitlines()) - self.assertTrue("__init__.py" not in ls().splitlines()) - self.assertRaises(OSError, local.cwd.chdir, "../non_exist1N9") - - def test_path(self): - self.assertFalse((local.cwd / "../non_exist1N9").exists()) - self.assertTrue((local.cwd / ".." / "plumbum").isdir()) - # traversal - found = False - for fn in local.cwd / ".." / "plumbum": - if fn.basename == "__init__.py": - self.assertTrue(fn.isfile()) - found = True - self.assertTrue(found) - # glob'ing - found = False - for fn in local.cwd / ".." // "*/*.rst": - if fn.basename == "index.rst": - found = True - self.assertTrue(found) - - def test_env(self): - self.assertTrue("PATH" in local.env) - self.assertFalse("FOOBAR72" in local.env) - self.assertRaises(ProcessExecutionError, local.python, "-c", "import os;os.environ['FOOBAR72']") - local.env["FOOBAR72"] = "spAm" - self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR72'])").splitlines(), ["spAm"]) - - with local.env(FOOBAR73 = 1889): - self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR73'])").splitlines(), ["1889"]) - with local.env(FOOBAR73 = 1778): - self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR73'])").splitlines(), ["1778"]) - self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR73'])").splitlines(), ["1889"]) - self.assertRaises(ProcessExecutionError, local.python, "-c", "import os;os.environ['FOOBAR73']") - - # path manipulation - self.assertRaises(CommandNotFound, local.which, "dummy-executable") - with local.env(): - local.env.path.insert(0, local.cwd / "not-in-path") - p = local.which("dummy-executable") - self.assertEqual(p, local.cwd / "not-in-path" / "dummy-executable") - - def test_local(self): - self.assertTrue("plumbum" in str(local.cwd)) - self.assertTrue("PATH" in local.env.getdict()) - self.assertEqual(local.path("foo"), os.path.join(os.getcwd(), "foo")) - local.which("ls") - local["ls"] - self.assertEqual(local.python("-c", "print ('hi there')").splitlines(), ["hi there"]) - - def test_piping(self): - from plumbum.cmd import ls, grep - chain = ls | grep["\\.py"] - self.assertTrue("test_local.py" in chain().splitlines()) - - chain = (ls["-a"] | grep["test"] | grep["local"]) - self.assertTrue("test_local.py" in chain().splitlines()) - - def test_redirection(self): - from plumbum.cmd import cat, ls, grep, rm - - chain = (ls | grep["\\.py"]) > "tmp.txt" - chain() - - chain2 = (cat < "tmp.txt") | grep["local"] - self.assertTrue("test_local.py" in chain2().splitlines()) - rm("tmp.txt") - - chain3 = (cat << "this is the\nworld of helloness and\nspam bar and eggs") | grep["hello"] - self.assertTrue("world of helloness and" in chain3().splitlines()) - - rc, _, err = (grep["-Zq5"] >= "tmp2.txt").run(["-Zq5"], retcode = None) - self.assertEqual(rc, 2) - self.assertFalse(err) - self.assertTrue("Usage" in (cat < "tmp2.txt")()) - rm("tmp2.txt") - - rc, out, _ = (grep["-Zq5"] >= ERROUT).run(["-Zq5"], retcode = None) - self.assertEqual(rc, 2) - self.assertTrue("Usage" in out) - - def test_popen(self): - from plumbum.cmd import ls - - p = ls.popen(["-a"]) - out, _ = p.communicate() - self.assertEqual(p.returncode, 0) - self.assertTrue("test_local.py" in out.decode(local.encoding).splitlines()) - - def test_run(self): - from plumbum.cmd import ls, grep - - rc, out, err = (ls | grep["non_exist1N9"]).run(retcode = 1) - self.assertEqual(rc, 1) - - def test_timeout(self): - from plumbum.cmd import sleep - self.assertRaises(ProcessTimedOut, sleep, 10, timeout = 5) - - - def test_modifiers(self): - from plumbum.cmd import ls, grep - f = (ls["-a"] | grep["\\.py"]) & BG - f.wait() - self.assertTrue("test_local.py" in f.stdout.splitlines()) - - (ls["-a"] | grep["local"]) & FG - - def test_session(self): - sh = local.session() - for _ in range(4): - _, out, _ = sh.run("ls -a") - self.assertTrue("test_local.py" in out.splitlines()) - - sh.run("cd ..") - sh.run("export FOO=17") - out = sh.run("echo $FOO")[1] - self.assertEqual(out.splitlines(), ["17"]) - - def test_quoting(self): - ssh = local["ssh"] - pwd = local["pwd"] - - cmd = ssh["localhost", "cd", "/usr", "&&", ssh["localhost", "cd", "/", "&&", - ssh["localhost", "cd", "/bin", "&&", pwd]]] - self.assertTrue("\"'&&'\"" in " ".join(cmd.formulate(0))) - - def test_tempdir(self): - from plumbum.cmd import cat - with local.tempdir() as dir: - self.assertTrue(dir.isdir()) - with open(str(dir / "test.txt"), "w") as f: - f.write("hello world") - with open(str(dir / "test.txt"), "r") as f: - self.assertEqual(f.read(), "hello world") - - self.assertFalse(dir.exists()) - - def test_read_write(self): - with local.tempdir() as tmp: - data = "hello world" - (tmp / "foo.txt").write(data) - self.assertEqual((tmp / "foo.txt").read(), data) - - -if __name__ == "__main__": - unittest.main() - +from __future__ import with_statement +import os +import unittest +import six +from plumbum import local, LocalPath, FG, BG, ERROUT +from plumbum import CommandNotFound, ProcessExecutionError, ProcessTimedOut + + +class LocalPathTest(unittest.TestCase): + def test_basename(self): + name = LocalPath("/some/long/path/to/file.txt").basename + self.assertTrue(isinstance(name, six.string_types)) + self.assertEqual("file.txt", str(name)) + + def test_dirname(self): + name = LocalPath("/some/long/path/to/file.txt").dirname + self.assertTrue(isinstance(name, LocalPath)) + self.assertEqual("/some/long/path/to", str(name)) + + # requires being run as root + def _test_chown(self): + path = LocalPath("/tmp/delme.txt") + path.delete() + path.write('test') + self.assertTrue('nobody' != path.owner) + self.assertTrue('nogroup' != path.group) + # chown group + path.chown(group='nogroup') + self.assertEqual('nogroup', path.group) + self.assertTrue('nobody' != path.owner) + # chown owner + path.chown('nobody') + self.assertEqual('nobody', path.owner) + # chown both / numerical ids + path.chown(uid=0, gid=0) + self.assertEqual('root', path.owner) + self.assertEqual('root', path.group) + # recursive + path.chown('root', recursive=True) + # set properties + path.owner = 'nobody' + self.assertEqual('nobody', path.owner) + path.group = 'nogroup' + self.assertEqual('nogroup', path.group) + path.delete() + + +class LocalMachineTest(unittest.TestCase): + def test_imports(self): + from plumbum.cmd import ls + self.assertTrue("test_local.py" in local["ls"]().splitlines()) + self.assertTrue("test_local.py" in ls().splitlines()) + + self.assertRaises(CommandNotFound, lambda: local["non_exist1N9"]) + + try: + from plumbum.cmd import non_exist1N9 #@UnresolvedImport @UnusedImport + except CommandNotFound: + pass + else: + self.fail("from plumbum.cmd import non_exist1N9") + + def test_cwd(self): + from plumbum.cmd import ls + self.assertEqual(local.cwd, os.getcwd()) + self.assertTrue("__init__.py" not in ls().splitlines()) + with local.cwd("../plumbum"): + self.assertTrue("__init__.py" in ls().splitlines()) + self.assertTrue("__init__.py" not in ls().splitlines()) + self.assertRaises(OSError, local.cwd.chdir, "../non_exist1N9") + + def test_path(self): + self.assertFalse((local.cwd / "../non_exist1N9").exists()) + self.assertTrue((local.cwd / ".." / "plumbum").isdir()) + # traversal + found = False + for fn in local.cwd / ".." / "plumbum": + if fn.basename == "__init__.py": + self.assertTrue(fn.isfile()) + found = True + self.assertTrue(found) + # glob'ing + found = False + for fn in local.cwd / ".." // "*/*.rst": + if fn.basename == "index.rst": + found = True + self.assertTrue(found) + + def test_env(self): + self.assertTrue("PATH" in local.env) + self.assertFalse("FOOBAR72" in local.env) + self.assertRaises(ProcessExecutionError, local.python, "-c", "import os;os.environ['FOOBAR72']") + local.env["FOOBAR72"] = "spAm" + self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR72'])").splitlines(), ["spAm"]) + + with local.env(FOOBAR73 = 1889): + self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR73'])").splitlines(), ["1889"]) + with local.env(FOOBAR73 = 1778): + self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR73'])").splitlines(), ["1778"]) + self.assertEqual(local.python("-c", "import os;print (os.environ['FOOBAR73'])").splitlines(), ["1889"]) + self.assertRaises(ProcessExecutionError, local.python, "-c", "import os;os.environ['FOOBAR73']") + + # path manipulation + self.assertRaises(CommandNotFound, local.which, "dummy-executable") + with local.env(): + local.env.path.insert(0, local.cwd / "not-in-path") + p = local.which("dummy-executable") + self.assertEqual(p, local.cwd / "not-in-path" / "dummy-executable") + + def test_local(self): + self.assertTrue("plumbum" in str(local.cwd)) + self.assertTrue("PATH" in local.env.getdict()) + self.assertEqual(local.path("foo"), os.path.join(os.getcwd(), "foo")) + local.which("ls") + local["ls"] + self.assertEqual(local.python("-c", "print ('hi there')").splitlines(), ["hi there"]) + + def test_piping(self): + from plumbum.cmd import ls, grep + chain = ls | grep["\\.py"] + self.assertTrue("test_local.py" in chain().splitlines()) + + chain = (ls["-a"] | grep["test"] | grep["local"]) + self.assertTrue("test_local.py" in chain().splitlines()) + + def test_redirection(self): + from plumbum.cmd import cat, ls, grep, rm + + chain = (ls | grep["\\.py"]) > "tmp.txt" + chain() + + chain2 = (cat < "tmp.txt") | grep["local"] + self.assertTrue("test_local.py" in chain2().splitlines()) + rm("tmp.txt") + + chain3 = (cat << "this is the\nworld of helloness and\nspam bar and eggs") | grep["hello"] + self.assertTrue("world of helloness and" in chain3().splitlines()) + + rc, _, err = (grep["-Zq5"] >= "tmp2.txt").run(["-Zq5"], retcode = None) + self.assertEqual(rc, 2) + self.assertFalse(err) + self.assertTrue("Usage" in (cat < "tmp2.txt")()) + rm("tmp2.txt") + + rc, out, _ = (grep["-Zq5"] >= ERROUT).run(["-Zq5"], retcode = None) + self.assertEqual(rc, 2) + self.assertTrue("Usage" in out) + + def test_popen(self): + from plumbum.cmd import ls + + p = ls.popen(["-a"]) + out, _ = p.communicate() + self.assertEqual(p.returncode, 0) + self.assertTrue("test_local.py" in out.decode(local.encoding).splitlines()) + + def test_run(self): + from plumbum.cmd import ls, grep + + rc, out, err = (ls | grep["non_exist1N9"]).run(retcode = 1) + self.assertEqual(rc, 1) + + def test_timeout(self): + from plumbum.cmd import sleep + self.assertRaises(ProcessTimedOut, sleep, 10, timeout = 5) + + + def test_modifiers(self): + from plumbum.cmd import ls, grep + f = (ls["-a"] | grep["\\.py"]) & BG + f.wait() + self.assertTrue("test_local.py" in f.stdout.splitlines()) + + (ls["-a"] | grep["local"]) & FG + + def test_session(self): + sh = local.session() + for _ in range(4): + _, out, _ = sh.run("ls -a") + self.assertTrue("test_local.py" in out.splitlines()) + + sh.run("cd ..") + sh.run("export FOO=17") + out = sh.run("echo $FOO")[1] + self.assertEqual(out.splitlines(), ["17"]) + + def test_quoting(self): + ssh = local["ssh"] + pwd = local["pwd"] + + cmd = ssh["localhost", "cd", "/usr", "&&", ssh["localhost", "cd", "/", "&&", + ssh["localhost", "cd", "/bin", "&&", pwd]]] + self.assertTrue("\"'&&'\"" in " ".join(cmd.formulate(0))) + + def test_tempdir(self): + from plumbum.cmd import cat + with local.tempdir() as dir: + self.assertTrue(dir.isdir()) + with open(str(dir / "test.txt"), "w") as f: + f.write("hello world") + with open(str(dir / "test.txt"), "r") as f: + self.assertEqual(f.read(), "hello world") + + self.assertFalse(dir.exists()) + + def test_read_write(self): + with local.tempdir() as tmp: + data = "hello world" + (tmp / "foo.txt").write(data) + self.assertEqual((tmp / "foo.txt").read(), data) + + +if __name__ == "__main__": + unittest.main() +