diff --git a/plumbum/local_machine.py b/plumbum/local_machine.py index 4772dedc6..80be3f163 100644 --- a/plumbum/local_machine.py +++ b/plumbum/local_machine.py @@ -1,20 +1,23 @@ from __future__ import with_statement -import sys 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.remote_path import RemotePath from plumbum.commands import CommandNotFound, ConcreteCommand from plumbum.session import ShellSession -from types import ModuleType -from tempfile import mkdtemp from plumbum.lib import _setdoc logger = logging.getLogger("plumbum.local") @@ -55,6 +58,26 @@ def basename(self): def dirname(self): return self.__class__(os.path.dirname(str(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) + + @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) + @_setdoc(Path) def join(self, other): if isinstance(other, RemotePath): @@ -133,6 +156,25 @@ def write(self, data): with self.open("w") as f: 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) + + class Workdir(LocalPath): """Working directory manipulator""" diff --git a/plumbum/path.py b/plumbum/path.py index cb23e2a4f..44652b699 100644 --- a/plumbum/path.py +++ b/plumbum/path.py @@ -1,12 +1,12 @@ class Path(object): """An abstraction over file system paths. This class is abstract, and the two implementations - are :class:`LocalPath ` and + are :class:`LocalPath ` and :class:`RemotePath `. """ - + __slots__ = [] CASE_SENSITIVE = True - + def __repr__(self): return "<%s %s>" % (self.__class__.__name__, str(self)) def __div__(self, other): @@ -44,7 +44,7 @@ def __hash__(self): def __nonzero__(self): return bool(str(self)) __bool__ = __nonzero__ - + def up(self, count = 1): """Go up in ``count`` directories (the default is 1)""" return self.join("../" * count) @@ -67,11 +67,19 @@ def basename(self): def dirname(self): """The dirname component of this path""" raise NotImplementedError() - + @property + def owner(self): + """The owner of leaf component of this path""" + raise NotImplementedError() + @property + def group(self): + """The group of leaf component of this path""" + raise NotImplementedError() + def _get_info(self): raise NotImplementedError() def join(self, *parts): - """Joins this path with any number of paths""" + """Joins this path with any number of paths""" raise NotImplementedError() def list(self): """Returns the files in this directory""" @@ -114,9 +122,6 @@ 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""" + raise NotImplementedError() diff --git a/plumbum/remote_path.py b/plumbum/remote_path.py index 3451d8912..fe20e95de 100644 --- a/plumbum/remote_path.py +++ b/plumbum/remote_path.py @@ -59,6 +59,27 @@ def dirname(self): return str(self) return self.__class__(self.remote, str(self).rsplit("/", 1)[0]) + @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) + + @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 _get_info(self): return (self.remote, self._path) @@ -160,3 +181,19 @@ def write(self, data): f.flush() f.seek(0) 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() + 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)) diff --git a/tests/test_local.py b/tests/test_local.py index 54d58e1cf..d5b945080 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -1,188 +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)) - - -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() +