diff options
author | Antoine Pitrou <solipsis@pitrou.net> | 2010-08-21 19:17:25 (GMT) |
---|---|---|
committer | Antoine Pitrou <solipsis@pitrou.net> | 2010-08-21 19:17:25 (GMT) |
commit | 3ebaed60b4779f89e5f5b82373c5a3bbc452813c (patch) | |
tree | dcd7d4e951de7873d5008d4956d22145c022c43f /Lib/test | |
parent | b7575ee2d3bf00c9e5bd1391f95eef6cdd59c53a (diff) | |
download | cpython-3ebaed60b4779f89e5f5b82373c5a3bbc452813c.zip cpython-3ebaed60b4779f89e5f5b82373c5a3bbc452813c.tar.gz cpython-3ebaed60b4779f89e5f5b82373c5a3bbc452813c.tar.bz2 |
Merged revisions 84239 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/branches/py3k
........
r84239 | antoine.pitrou | 2010-08-21 21:09:32 +0200 (sam., 21 août 2010) | 4 lines
Issue #9617: Signals received during a low-level write operation aren't
ignored by the buffered IO layer anymore.
........
Diffstat (limited to 'Lib/test')
-rw-r--r-- | Lib/test/test_io.py | 72 |
1 files changed, 72 insertions, 0 deletions
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index f25f710..8686f22 100644 --- a/Lib/test/test_io.py +++ b/Lib/test/test_io.py @@ -30,6 +30,8 @@ import random import unittest import weakref import abc +import signal +import errno from itertools import cycle, count from collections import deque from test import test_support as support @@ -2471,6 +2473,75 @@ class CMiscIOTest(MiscIOTest): class PyMiscIOTest(MiscIOTest): io = pyio + +@unittest.skipIf(os.name == 'nt', 'POSIX signals required for this test.') +class SignalsTest(unittest.TestCase): + + def setUp(self): + self.oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt) + + def tearDown(self): + signal.signal(signal.SIGALRM, self.oldalrm) + + def alarm_interrupt(self, sig, frame): + 1/0 + + @unittest.skipUnless(threading, 'Threading required for this test.') + def check_interrupted_write(self, item, bytes, **fdopen_kwargs): + """Check that a partial write, when it gets interrupted, properly + invokes the signal handler.""" + read_results = [] + def _read(): + s = os.read(r, 1) + read_results.append(s) + t = threading.Thread(target=_read) + t.daemon = True + r, w = os.pipe() + try: + wio = self.io.open(w, **fdopen_kwargs) + t.start() + signal.alarm(1) + # Fill the pipe enough that the write will be blocking. + # It will be interrupted by the timer armed above. Since the + # other thread has read one byte, the low-level write will + # return with a successful (partial) result rather than an EINTR. + # The buffered IO layer must check for pending signal + # handlers, which in this case will invoke alarm_interrupt(). + self.assertRaises(ZeroDivisionError, + wio.write, item * (1024 * 1024)) + t.join() + # We got one byte, get another one and check that it isn't a + # repeat of the first one. + read_results.append(os.read(r, 1)) + self.assertEqual(read_results, [bytes[0:1], bytes[1:2]]) + finally: + os.close(w) + os.close(r) + # This is deliberate. If we didn't close the file descriptor + # before closing wio, wio would try to flush its internal + # buffer, and block again. + try: + wio.close() + except IOError as e: + if e.errno != errno.EBADF: + raise + + def test_interrupted_write_unbuffered(self): + self.check_interrupted_write(b"xy", b"xy", mode="wb", buffering=0) + + def test_interrupted_write_buffered(self): + self.check_interrupted_write(b"xy", b"xy", mode="wb") + + def test_interrupted_write_text(self): + self.check_interrupted_write("xy", b"xy", mode="w", encoding="ascii") + +class CSignalsTest(SignalsTest): + io = io + +class PySignalsTest(SignalsTest): + io = pyio + + def test_main(): tests = (CIOTest, PyIOTest, CBufferedReaderTest, PyBufferedReaderTest, @@ -2481,6 +2552,7 @@ def test_main(): CIncrementalNewlineDecoderTest, PyIncrementalNewlineDecoderTest, CTextIOWrapperTest, PyTextIOWrapperTest, CMiscIOTest, PyMiscIOTest, + CSignalsTest, PySignalsTest, ) # Put the namespaces of the IO module we are testing and some useful mock |