Merge.

3 files changed, 139 insertions, 8 deletions

diff --git a/Lib/test/test_re.py b/Lib/test/test_re.py
index 011fba9..e937c85 100644
--- a/Lib/test/test_re.py
+++ b/Lib/test/test_re.py
@@ -1203,16 +1203,33 @@ class ReTests(unittest.TestCase):
                 self.assertEqual(m.group(2), "y")
 
     def test_debug_flag(self):
+        pat = r'(\.)(?:[ch]|py)(?(1)$|: )'
         with captured_stdout() as out:
-            re.compile('foo', re.DEBUG)
-        self.assertEqual(out.getvalue().splitlines(),
-                         ['literal 102 ', 'literal 111 ', 'literal 111 '])
+            re.compile(pat, re.DEBUG)
+        dump = '''\
+subpattern 1
+  literal 46
+subpattern None
+  branch
+    in
+      literal 99
+      literal 104
+  or
+    literal 112
+    literal 121
+subpattern None
+  groupref_exists 1
+    at at_end
+  else
+    literal 58
+    literal 32
+'''
+        self.assertEqual(out.getvalue(), dump)
         # Debug output is output again even a second time (bypassing
         # the cache -- issue #20426).
         with captured_stdout() as out:
-            re.compile('foo', re.DEBUG)
-        self.assertEqual(out.getvalue().splitlines(),
-                         ['literal 102 ', 'literal 111 ', 'literal 111 '])
+            re.compile(pat, re.DEBUG)
+        self.assertEqual(out.getvalue(), dump)
 
     def test_keyword_parameters(self):
         # Issue #20283: Accepting the string keyword parameter.
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index d0ab718..9616da0 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -1008,6 +1008,39 @@ class ProcessTestCase(BaseTestCase):
         p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None)
         self.assertEqual(p.wait(), 0)
 
+    def _test_bufsize_equal_one(self, line, expected, universal_newlines):
+        # subprocess may deadlock with bufsize=1, see issue #21332
+        with subprocess.Popen([sys.executable, "-c", "import sys;"
+                               "sys.stdout.write(sys.stdin.readline());"
+                               "sys.stdout.flush()"],
+                              stdin=subprocess.PIPE,
+                              stdout=subprocess.PIPE,
+                              stderr=subprocess.DEVNULL,
+                              bufsize=1,
+                              universal_newlines=universal_newlines) as p:
+            p.stdin.write(line) # expect that it flushes the line in text mode
+            os.close(p.stdin.fileno()) # close it without flushing the buffer
+            read_line = p.stdout.readline()
+            try:
+                p.stdin.close()
+            except OSError:
+                pass
+            p.stdin = None
+        self.assertEqual(p.returncode, 0)
+        self.assertEqual(read_line, expected)
+
+    def test_bufsize_equal_one_text_mode(self):
+        # line is flushed in text mode with bufsize=1.
+        # we should get the full line in return
+        line = "line\n"
+        self._test_bufsize_equal_one(line, line, universal_newlines=True)
+
+    def test_bufsize_equal_one_binary_mode(self):
+        # line is not flushed in binary mode with bufsize=1.
+        # we should get empty response
+        line = b'line' + os.linesep.encode() # assume ascii-based locale
+        self._test_bufsize_equal_one(line, b'', universal_newlines=False)
+
     def test_leaking_fds_on_error(self):
         # see bug #5179: Popen leaks file descriptors to PIPEs if
         # the child fails to execute; this will eventually exhaust
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
index 7170f60..98f01ee 100644
--- a/Lib/test/test_threading.py
+++ b/Lib/test/test_threading.py
@@ -4,7 +4,7 @@ Tests for the threading module.
 
 import test.support
 from test.support import verbose, strip_python_stderr, import_module, cpython_only
-from test.script_helper import assert_python_ok
+from test.script_helper import assert_python_ok, assert_python_failure
 
 import random
 import re
@@ -15,7 +15,6 @@ import time
 import unittest
 import weakref
 import os
-from test.script_helper import assert_python_ok, assert_python_failure
 import subprocess
 
 from test import lock_tests
@@ -962,6 +961,88 @@ class ThreadingExceptionTests(BaseTestCase):
         self.assertEqual(p.returncode, 0, "Unexpected error: " + stderr.decode())
         self.assertEqual(data, expected_output)
 
+    def test_print_exception(self):
+        script = r"""if True:
+            import threading
+            import time
+
+            running = False
+            def run():
+                global running
+                running = True
+                while running:
+                    time.sleep(0.01)
+                1/0
+            t = threading.Thread(target=run)
+            t.start()
+            while not running:
+                time.sleep(0.01)
+            running = False
+            t.join()
+            """
+        rc, out, err = assert_python_ok("-c", script)
+        self.assertEqual(out, b'')
+        err = err.decode()
+        self.assertIn("Exception in thread", err)
+        self.assertIn("Traceback (most recent call last):", err)
+        self.assertIn("ZeroDivisionError", err)
+        self.assertNotIn("Unhandled exception", err)
+
+    def test_print_exception_stderr_is_none_1(self):
+        script = r"""if True:
+            import sys
+            import threading
+            import time
+
+            running = False
+            def run():
+                global running
+                running = True
+                while running:
+                    time.sleep(0.01)
+                1/0
+            t = threading.Thread(target=run)
+            t.start()
+            while not running:
+                time.sleep(0.01)
+            sys.stderr = None
+            running = False
+            t.join()
+            """
+        rc, out, err = assert_python_ok("-c", script)
+        self.assertEqual(out, b'')
+        err = err.decode()
+        self.assertIn("Exception in thread", err)
+        self.assertIn("Traceback (most recent call last):", err)
+        self.assertIn("ZeroDivisionError", err)
+        self.assertNotIn("Unhandled exception", err)
+
+    def test_print_exception_stderr_is_none_2(self):
+        script = r"""if True:
+            import sys
+            import threading
+            import time
+
+            running = False
+            def run():
+                global running
+                running = True
+                while running:
+                    time.sleep(0.01)
+                1/0
+            sys.stderr = None
+            t = threading.Thread(target=run)
+            t.start()
+            while not running:
+                time.sleep(0.01)
+            running = False
+            t.join()
+            """
+        rc, out, err = assert_python_ok("-c", script)
+        self.assertEqual(out, b'')
+        self.assertNotIn("Unhandled exception", err.decode())
+
+
 class TimerTests(BaseTestCase):
 
     def setUp(self):