Repository: amoffat/sh
Pull request #542 (2020-09-04 20:52:41)

Fix/pytest
The `_disable_whitelist` test backdoor was bleeding shared state into tests. It only manifested in pytest because of the order pytest chose to run tests. Removed the backdoor and just made the required class "whitelisted"


Patch:
diff --git a/sh.py b/sh.py
index 4acebf8..d7038d8 100644
--- a/sh.py
+++ b/sh.py
@@ -2900,8 +2900,8 @@ def determine_how_to_feed_output(handler, encoding, decode_errors):
         try:
             handler = int(handler)
         except (ValueError, TypeError):
-            process = lambda chunk: False  # noqa: E731
-            finish = lambda: None  # noqa: E731
+            def process(chunk): return False  # noqa: E731
+            def finish(): return None  # noqa: E731
         else:
             process, finish = get_fd_chunk_consumer(handler)
 
@@ -2915,14 +2915,14 @@ def get_fd_chunk_consumer(handler):
 
 def get_file_chunk_consumer(handler):
     if getattr(handler, "encoding", None):
-        encode = lambda chunk: chunk.decode(handler.encoding)  # noqa: E731
+        def encode(chunk): return chunk.decode(handler.encoding)  # noqa: E731
     else:
-        encode = lambda chunk: chunk  # noqa: E731
+        def encode(chunk): return chunk  # noqa: E731
 
     if hasattr(handler, "flush"):
         flush = handler.flush
     else:
-        flush = lambda: None  # noqa: E731
+        def flush(): return None  # noqa: E731
 
     def process(chunk):
         handler.write(encode(chunk))
@@ -3246,6 +3246,7 @@ class Environment(dict):
         "SignalException",
         "ForkException",
         "TimeoutException",
+        "StreamBufferer",
         "__project_url__",
         "__version__",
         "__file__",
@@ -3265,23 +3266,15 @@ class Environment(dict):
         super(dict, self).__init__()
         self.globs = globs
         self.baked_args = baked_args or {}
-        self.disable_whitelist = False
 
     def __getitem__(self, k):
-        # if we first import "_disable_whitelist" from sh, we can import
-        # anything defined in the global scope of sh.py.  this is useful for our
-        # tests
-        if k == "_disable_whitelist":
-            self.disable_whitelist = True
-            return None
-
         if k == 'args':
             # Let the deprecated '_args' context manager be imported as 'args'
             k = '_args'
 
-        # we're trying to import something real (maybe), see if it's in our
-        # global scope
-        if k in self.whitelist or self.disable_whitelist:
+        # if we're trying to import something real, see if it's in our global scope.
+        # what defines "real" is that it's in our whitelist
+        if k in self.whitelist:
             return self.globs[k]
 
         # somebody tried to be funny and do "from sh import *"
@@ -3459,15 +3452,15 @@ def ssh(orig):  # pragma: no cover
         prompt = "Please enter SSH password: "
 
         if prompt_match is None:
-            prompt_match = lambda content: content.cur_line.endswith("password: ")  # noqa: E731
+            def prompt_match(content): return content.cur_line.endswith("password: ")  # noqa: E731
 
         if password is None:
-            pass_getter = lambda: getpass.getpass(prompt=prompt)  # noqa: E731
+            def pass_getter(): return getpass.getpass(prompt=prompt)  # noqa: E731
         else:
-            pass_getter = lambda: password.rstrip("\n")  # noqa: E731
+            def pass_getter(): return password.rstrip("\n")  # noqa: E731
 
         if login_success is None:
-            login_success = lambda content: True  # noqa: E731
+            def login_success(content): return True  # noqa: E731
 
         kwargs["_out"] = SSHInteract(prompt_match, pass_getter, real_out_handler, login_success)
         return a, kwargs


Test patch:
diff --git a/test.py b/test.py
index 142fcc0..a4b9915 100644
--- a/test.py
+++ b/test.py
@@ -315,7 +315,7 @@ sys.stdout.write("a" * 1000)
 sys.stderr.write("b" * 1000)
 exit(1)
 """)
-        self.assertRaises(sh.ErrorReturnCode, python, py.name)
+        self.assertRaises(sh.ErrorReturnCode_1, python, py.name)
 
     def test_number_arg(self):
         py = create_tmp_test("""
@@ -341,11 +341,11 @@ sys.stdout.write("no hang")
         self.assertEqual(out, "no hang")
 
     def test_exit_code(self):
-        from sh import ErrorReturnCode
+        from sh import ErrorReturnCode_3
         py = create_tmp_test("""
 exit(3)
 """)
-        self.assertRaises(ErrorReturnCode, python, py.name)
+        self.assertRaises(ErrorReturnCode_3, python, py.name)
 
     def test_patched_glob(self):
         from glob import glob
@@ -371,7 +371,7 @@ print(sys.argv[1:])
         self.assertEqual(out, "['*.faowjefoajweofj']")
 
     def test_exit_code_with_hasattr(self):
-        from sh import ErrorReturnCode
+        from sh import ErrorReturnCode_3
         py = create_tmp_test("""
 exit(3)
 """)
@@ -383,16 +383,16 @@ exit(3)
             list(out)
             self.assertEqual(out.exit_code, 3)
             self.fail("Command exited with error, but no exception thrown")
-        except ErrorReturnCode:
+        except ErrorReturnCode_3:
             pass
 
     def test_exit_code_from_exception(self):
-        from sh import ErrorReturnCode
+        from sh import ErrorReturnCode_3
         py = create_tmp_test("""
 exit(3)
 """)
 
-        self.assertRaises(ErrorReturnCode, python, py.name)
+        self.assertRaises(ErrorReturnCode_3, python, py.name)
 
         try:
             python(py.name)
@@ -3043,7 +3043,6 @@ time.sleep(2)
 
 class StreamBuffererTests(unittest.TestCase):
     def test_unbuffered(self):
-        from sh import _disable_whitelist  # noqa: F401
         from sh import StreamBufferer
         b = StreamBufferer(0)
 
@@ -3053,7 +3052,6 @@ class StreamBuffererTests(unittest.TestCase):
         self.assertEqual(b.flush(), b"")
 
     def test_newline_buffered(self):
-        from sh import _disable_whitelist  # noqa: F401
         from sh import StreamBufferer
         b = StreamBufferer(1)
 
@@ -3062,7 +3060,6 @@ class StreamBuffererTests(unittest.TestCase):
         self.assertEqual(b.flush(), b"four")
 
     def test_chunk_buffered(self):
-        from sh import _disable_whitelist  # noqa: F401
         from sh import StreamBufferer
         b = StreamBufferer(10)