diff options
Diffstat (limited to 'Lib/test')
| -rw-r--r-- | Lib/test/test_re.py | 29 | ||||
| -rw-r--r-- | Lib/test/test_subprocess.py | 33 | ||||
| -rw-r--r-- | Lib/test/test_threading.py | 85 |
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): |