Fixing tests on Mac

plumbum/lib.py

@@ -1,4 +1,5 @@
 import sys
+from contextlib import contextmanager
 from abc import ABCMeta
 import inspect
 
@@ -70,6 +71,51 @@ def u(s):
         def get_method_function(m):
             return m.im_func
 
+# Try/except fails because io has the wrong StringIO in Python2
+# You'll get str/unicode errors
+if six.PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
+
+def ensure_skipIf(unittest):
+    """
+    This will ensure that unittest has skipIf. Call like::
+
+        import unittest
+        ensure_skipIf(unittest)
+    """
+
+    if not hasattr(unittest, "skipIf"):
+        import logging
+        import functools
+        def skipIf(condition, reason):
+            def deco(func):
+                if condition:
+                    return func
+                else:
+                    @functools.wraps(func)
+                    def wrapper(*args, **kwargs):
+                        logging.warn("skipping test: "+reason)
+                    return wrapper
+            return deco
+        unittest.skipIf = skipIf
+
+@contextmanager
+def captured_stdout(stdin = ""):
+    """
+    Captures stdout (similar to the redirect_stdout in Python 3.4+, but with slightly different arguments)
+    """
+    prevstdin = sys.stdin
+    prevstdout = sys.stdout
+    sys.stdin = StringIO(six.u(stdin))
+    sys.stdout = StringIO()
+    try:
+        yield sys.stdout
+    finally:
+        sys.stdin = prevstdin
+        sys.stdout = prevstdout
+
 class StaticProperty(object):
     """This acts like a static property, allowing access via class or object.
     This is a non-data descriptor."""

tests/test_cli.py

@@ -1,32 +1,11 @@
 import sys
 import unittest
 from plumbum import cli
-from contextlib import contextmanager
 import time
 
 from plumbum import cli, local
 from plumbum.cli.terminal import ask, choose, hexdump, Progress
-from plumbum.lib import six
-
-# string/unicode issues
-if six.PY3:
-    from io import StringIO
-else:
-    from StringIO import StringIO
-
-
-@contextmanager
-def captured_stdout(stdin = ""):
-    prevstdin = sys.stdin
-    prevstdout = sys.stdout
-    sys.stdin = StringIO(six.u(stdin))
-    sys.stdout = StringIO()
-    try:
-        yield sys.stdout
-    finally:
-        sys.stdin = prevstdin
-        sys.stdout = prevstdout
-
+from plumbum.lib import six, captured_stdout, StringIO
 
 class TestApp(cli.Application):
     @cli.switch(["a"])

tests/test_local.py

@@ -4,27 +4,13 @@
 import signal
 import time
 from plumbum import local, LocalPath, FG, BG, TF, RETCODE, ERROUT
-from plumbum.lib import six
+from plumbum.lib import six, ensure_skipIf
 from plumbum import CommandNotFound, ProcessExecutionError, ProcessTimedOut
 from plumbum.fs.atomic import AtomicFile, AtomicCounterFile, PidFile
 from plumbum.path import RelativePath
 import plumbum
 
-
-if not hasattr(unittest, "skipIf"):
-    import logging
-    import functools
-    def skipIf(cond, msg = None):
-        def deco(func):
-            if cond:
-                return func
-            else:
-                @functools.wraps(func)
-                def wrapper(*args, **kwargs):
-                    logging.warn("skipping test")
-                return wrapper
-        return deco
-    unittest.skipIf = skipIf
+ensure_skipIf(unittest)
 
 class LocalPathTest(unittest.TestCase):
     def test_basename(self):
@@ -211,12 +197,12 @@ def test_redirection(self):
         rc, _, err = (grep["-Zq5"] >= "tmp2.txt").run(["-Zq5"], retcode = None)
         self.assertEqual(rc, 2)
         self.assertFalse(err)
-        self.assertTrue("Usage" in (cat < "tmp2.txt")())
+        self.assertTrue("usage" in (cat < "tmp2.txt")().lower())
         rm("tmp2.txt")
 
         rc, out, _ = (grep["-Zq5"] >= ERROUT).run(["-Zq5"], retcode = None)
         self.assertEqual(rc, 2)
-        self.assertTrue("Usage" in out)
+        self.assertTrue("usage" in out.lower())
 
     def test_popen(self):
         from plumbum.cmd import ls
@@ -240,7 +226,8 @@ def test_iter_lines_timeout(self):
         from plumbum.cmd import ping
 
         try:
-            for i, (out, err) in enumerate(ping["127.0.0.1", "-i", 0.5].popen().iter_lines(timeout=2)):
+            # Order is important on mac
+            for i, (out, err) in enumerate(ping["-i", 0.5, "127.0.0.1"].popen().iter_lines(timeout=2)):
                 print("out:", out)
                 print("err:", err)
         except ProcessTimedOut:
@@ -257,7 +244,8 @@ def test_iter_lines_error(self):
             self.assertEqual(i, 1)
         except ProcessExecutionError:
             ex = sys.exc_info()[1]
-            self.assertTrue(ex.stderr.startswith("/bin/ls: unrecognized option '--bla'"))
+            self.assertTrue(ex.stderr.startswith("/bin/ls: unrecognized option '--bla'")
+                            or ex.stderr.startswith("/bin/ls: illegal option -- -"))
         else:
             self.fail("Expected an execution error")
 
@@ -487,8 +475,10 @@ def test_bound_env(self):
         except CommandNotFound:
             self.skipTest("printenv is missing")
         with local.env(FOO = "hello"):
-            self.assertEqual(printenv.with_env(BAR = "world")("FOO", "BAR"), "hello\nworld\n")
-            self.assertEqual(printenv.with_env(FOO = "sea", BAR = "world")("FOO", "BAR"), "sea\nworld\n")
+            self.assertEqual(printenv.with_env(BAR = "world")("FOO"), "hello\n")
+            self.assertEqual(printenv.with_env(BAR = "world")("BAR"), "world\n")
+            self.assertEqual(printenv.with_env(FOO = "sea", BAR = "world")("FOO"), "sea\n")
+            self.assertEqual(printenv("FOO"), "hello\n")
 
     def test_nesting_lists_as_argv(self):
         from plumbum.cmd import ls

tests/test_remote.py

@@ -6,29 +6,16 @@
 import logging
 from plumbum import RemotePath, SshMachine, ProcessExecutionError, local, ProcessTimedOut, NOHUP
 from plumbum import CommandNotFound
-from plumbum.lib import six
+from plumbum.lib import six, ensure_skipIf
 
+ensure_skipIf(unittest)
 
 #TEST_HOST = "192.168.1.143"
 TEST_HOST = "127.0.0.1"
 if TEST_HOST not in ("::1", "127.0.0.1", "localhost"):
     import plumbum
     plumbum.local.env.path.append("c:\\Program Files\\Git\\bin")
 
-if not hasattr(unittest, "skipIf"):
-    import functools
-    def skipIf(cond, msg = None):
-        def deco(func):
-            if cond:
-                return func
-            else:
-                @functools.wraps(func)
-                def wrapper(*args, **kwargs):
-                    logging.warn("skipping test")
-                return wrapper
-        return deco
-    unittest.skipIf = skipIf
-
 class RemotePathTest(unittest.TestCase):
     def _connect(self):
         return SshMachine(TEST_HOST)
@@ -101,7 +88,8 @@ def test_basic(self):
             r_ls = rem["ls"]
             r_grep = rem["grep"]
 
-            self.assertTrue(".bashrc" in r_ls("-a").splitlines())
+            lines = r_ls("-a").splitlines()
+            self.assertTrue(".bashrc" in lines or ".bash_profile" in lines)
             with rem.cwd(os.path.dirname(os.path.abspath(__file__))):
                 cmd = r_ssh["localhost", "cd", rem.cwd, "&&", r_ls, "|", r_grep["\\.py"]]
                 self.assertTrue("'|'" in str(cmd))
@@ -130,7 +118,7 @@ def test_session(self):
             sh = rem.session()
             for _ in range(4):
                 _, out, _ = sh.run("ls -a")
-                self.assertTrue(".bashrc" in out)
+                self.assertTrue(".bashrc" in out or ".bash_profile" in out)
 
     def test_env(self):
         with self._connect() as rem:
@@ -167,7 +155,7 @@ def test_contains(self):
     def test_iter_lines_timeout(self):
         with self._connect() as rem:
             try:
-                for i, (out, err) in enumerate(rem["ping"]["127.0.0.1", "-i", 0.5].popen().iter_lines(timeout=2)):
+                for i, (out, err) in enumerate(rem["ping"]["-i", 0.5, "127.0.0.1"].popen().iter_lines(timeout=2)):
                     print("out:", out)
                     print("err:", err)
             except NotImplementedError:
@@ -189,7 +177,7 @@ def test_iter_lines_error(self):
                 self.assertEqual(i, 1)
             except ProcessExecutionError:
                 ex = sys.exc_info()[1]
-                self.assertTrue(ex.stderr.startswith("/bin/ls: unrecognized option '--bla'"))
+                self.assertTrue(ex.stderr.startswith("/bin/ls: "))
             else:
                 self.fail("Expected an execution error")
 
@@ -246,9 +234,12 @@ def test_bound_env(self):
         with self._connect() as rem:
             printenv = rem["printenv"]
             with rem.env(FOO = "hello"):
-                self.assertEqual(printenv.with_env(BAR = "world")("FOO", "BAR"), "hello\nworld\n")
-                self.assertEqual(printenv.with_env(FOO = "sea", BAR = "world")("FOO", "BAR"), "sea\nworld\n")
+                self.assertEqual(printenv.with_env(BAR = "world")("FOO"), "hello\n")
+                self.assertEqual(printenv.with_env(BAR = "world")("BAR"), "world\n")
+                self.assertEqual(printenv.with_env(FOO = "sea", BAR = "world")("FOO"), "sea\n")
+                self.assertEqual(printenv.with_env(FOO = "sea", BAR = "world")("BAR"), "world\n")
 
+    @unittest.skipIf('useradd' not in local, "System does not have useradd (Mac?)")
     def test_sshpass(self):
         with local.as_root():
             local["useradd"]("-m", "-b", "/tmp", "testuser")