traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Issue #25122: sync test_eintr with Python 3.6 

* test_eintr: support verbose mode, don't redirect eintr_tester output into
  a pipe
* eintr_tester: replace os.fork() with subprocess to have a cleaner child
  process (ex: don't inherit setitimer())
* eintr_tester: kill the process if the unit test fails
* test_open/test_os_open(): write support.PIPE_MAX_SIZE bytes instead of
  support.PIPE_MAX_SIZE*3 bytes
This commit is contained in:
Victor Stinner 2015-09-18 11:29:16 +02:00
parent cd6e69439c
commit 4b352171d2
2 changed files with 198 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

283
Lib/test/eintrdata/eintr_tester.py
View File

@ -8,16 +8,29 @@ Signals are generated in-process using setitimer(ITIMER_REAL), which allows
sub-second periodicity (contrarily to signal()).
"""
import contextlib
import io
import os
import select
import signal
import socket
import subprocess
import sys
import time
import unittest
from test import support
@contextlib.contextmanager
def kill_on_error(proc):
"""Context manager killing the subprocess if a Python exception is raised."""
with proc:
try:
yield proc
except:
proc.kill()
raise
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
class EINTRBaseTest(unittest.TestCase):
@ -28,7 +41,7 @@ class EINTRBaseTest(unittest.TestCase):
# signal delivery periodicity
signal_period = 0.1
# default sleep time for tests - should obviously have:
# sleep_time > signal_period
# sleep_time > signal_period
sleep_time = 0.2
@classmethod
@ -51,18 +64,22 @@ class EINTRBaseTest(unittest.TestCase):
# default sleep time
time.sleep(cls.sleep_time)
def subprocess(self, *args, **kw):
cmd_args = (sys.executable, '-c') + args
return subprocess.Popen(cmd_args, **kw)
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
class OSEINTRTest(EINTRBaseTest):
""" EINTR tests for the os module. """
def new_sleep_process(self):
code = 'import time; time.sleep(%r)' % self.sleep_time
return self.subprocess(code)
def _test_wait_multiple(self, wait_func):
num = 3
for _ in range(num):
pid = os.fork()
if pid == 0:
self._sleep()
os._exit(0)
processes = [self.new_sleep_process() for _ in range(num)]
for _ in range(num):
wait_func()
@ -74,12 +91,8 @@ class OSEINTRTest(EINTRBaseTest):
self._test_wait_multiple(lambda: os.wait3(0))
def _test_wait_single(self, wait_func):
pid = os.fork()
if pid == 0:
self._sleep()
os._exit(0)
else:
wait_func(pid)
proc = self.new_sleep_process()
wait_func(proc.pid)
def test_waitpid(self):
self._test_wait_single(lambda pid: os.waitpid(pid, 0))
@ -97,19 +110,25 @@ class OSEINTRTest(EINTRBaseTest):
# atomic
datas = [b"hello", b"world", b"spam"]
pid = os.fork()
if pid == 0:
os.close(rd)
for data in datas:
# let the parent block on read()
self._sleep()
os.write(wr, data)
os._exit(0)
else:
self.addCleanup(os.waitpid, pid, 0)
code = '\n'.join((
'import os, sys, time',
'',
'wr = int(sys.argv[1])',
'datas = %r' % datas,
'sleep_time = %r' % self.sleep_time,
'',
'for data in datas:',
' # let the parent block on read()',
' time.sleep(sleep_time)',
' os.write(wr, data)',
))
proc = self.subprocess(code, str(wr), pass_fds=[wr])
with kill_on_error(proc):
os.close(wr)
for data in datas:
self.assertEqual(data, os.read(rd, len(data)))
self.assertEqual(proc.wait(), 0)
def test_write(self):
rd, wr = os.pipe()
@ -117,25 +136,37 @@ class OSEINTRTest(EINTRBaseTest):
# rd closed explicitly by parent
# we must write enough data for the write() to block
data = b"xyz" * support.PIPE_MAX_SIZE
data = b"x" * support.PIPE_MAX_SIZE
pid = os.fork()
if pid == 0:
os.close(wr)
read_data = io.BytesIO()
# let the parent block on write()
self._sleep()
while len(read_data.getvalue()) < len(data):
chunk = os.read(rd, 2 * len(data))
read_data.write(chunk)
self.assertEqual(read_data.getvalue(), data)
os._exit(0)
else:
code = '\n'.join((
'import io, os, sys, time',
'',
'rd = int(sys.argv[1])',
'sleep_time = %r' % self.sleep_time,
'data = b"x" * %s' % support.PIPE_MAX_SIZE,
'data_len = len(data)',
'',
'# let the parent block on write()',
'time.sleep(sleep_time)',
'',
'read_data = io.BytesIO()',
'while len(read_data.getvalue()) < data_len:',
' chunk = os.read(rd, 2 * data_len)',
' read_data.write(chunk)',
'',
'value = read_data.getvalue()',
'if value != data:',
' raise Exception("read error: %s vs %s bytes"',
' % (len(value), data_len))',
))
proc = self.subprocess(code, str(rd), pass_fds=[rd])
with kill_on_error(proc):
os.close(rd)
written = 0
while written < len(data):
written += os.write(wr, memoryview(data)[written:])
self.assertEqual(0, os.waitpid(pid, 0)[1])
self.assertEqual(proc.wait(), 0)
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
@ -151,19 +182,32 @@ class SocketEINTRTest(EINTRBaseTest):
# single-byte payload guard us against partial recv
datas = [b"x", b"y", b"z"]
pid = os.fork()
if pid == 0:
rd.close()
for data in datas:
# let the parent block on recv()
self._sleep()
wr.sendall(data)
os._exit(0)
else:
self.addCleanup(os.waitpid, pid, 0)
code = '\n'.join((
'import os, socket, sys, time',
'',
'fd = int(sys.argv[1])',
'family = %s' % int(wr.family),
'sock_type = %s' % int(wr.type),
'datas = %r' % datas,
'sleep_time = %r' % self.sleep_time,
'',
'wr = socket.fromfd(fd, family, sock_type)',
'os.close(fd)',
'',
'with wr:',
' for data in datas:',
' # let the parent block on recv()',
' time.sleep(sleep_time)',
' wr.sendall(data)',
))
fd = wr.fileno()
proc = self.subprocess(code, str(fd), pass_fds=[fd])
with kill_on_error(proc):
wr.close()
for data in datas:
self.assertEqual(data, recv_func(rd, len(data)))
self.assertEqual(proc.wait(), 0)
def test_recv(self):
self._test_recv(socket.socket.recv)
@ -180,25 +224,43 @@ class SocketEINTRTest(EINTRBaseTest):
# we must send enough data for the send() to block
data = b"xyz" * (support.SOCK_MAX_SIZE // 3)
pid = os.fork()
if pid == 0:
wr.close()
# let the parent block on send()
self._sleep()
received_data = bytearray(len(data))
n = 0
while n < len(data):
n += rd.recv_into(memoryview(received_data)[n:])
self.assertEqual(received_data, data)
os._exit(0)
else:
code = '\n'.join((
'import os, socket, sys, time',
'',
'fd = int(sys.argv[1])',
'family = %s' % int(rd.family),
'sock_type = %s' % int(rd.type),
'sleep_time = %r' % self.sleep_time,
'data = b"xyz" * %s' % (support.SOCK_MAX_SIZE // 3),
'data_len = len(data)',
'',
'rd = socket.fromfd(fd, family, sock_type)',
'os.close(fd)',
'',
'with rd:',
' # let the parent block on send()',
' time.sleep(sleep_time)',
'',
' received_data = bytearray(data_len)',
' n = 0',
' while n < data_len:',
' n += rd.recv_into(memoryview(received_data)[n:])',
'',
'if received_data != data:',
' raise Exception("recv error: %s vs %s bytes"',
' % (len(received_data), data_len))',
))
fd = rd.fileno()
proc = self.subprocess(code, str(fd), pass_fds=[fd])
with kill_on_error(proc):
rd.close()
written = 0
while written < len(data):
sent = send_func(wr, memoryview(data)[written:])
# sendall() returns None
written += len(data) if sent is None else sent
self.assertEqual(0, os.waitpid(pid, 0)[1])
self.assertEqual(proc.wait(), 0)
def test_send(self):
self._test_send(socket.socket.send)
@ -215,46 +277,75 @@ class SocketEINTRTest(EINTRBaseTest):
self.addCleanup(sock.close)
sock.bind((support.HOST, 0))
_, port = sock.getsockname()
port = sock.getsockname()[1]
sock.listen()
pid = os.fork()
if pid == 0:
# let parent block on accept()
self._sleep()
with socket.create_connection((support.HOST, port)):
self._sleep()
os._exit(0)
else:
self.addCleanup(os.waitpid, pid, 0)
code = '\n'.join((
'import socket, time',
'',
'host = %r' % support.HOST,
'port = %s' % port,
'sleep_time = %r' % self.sleep_time,
'',
'# let parent block on accept()',
'time.sleep(sleep_time)',
'with socket.create_connection((host, port)):',
' time.sleep(sleep_time)',
))
proc = self.subprocess(code)
with kill_on_error(proc):
client_sock, _ = sock.accept()
client_sock.close()
self.assertEqual(proc.wait(), 0)
# Issue #25122: There is a race condition in the FreeBSD kernel on
# handling signals in the FIFO device. Skip the test until the bug is
# fixed in the kernel. Bet that the bug will be fixed in FreeBSD 11.
# https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=203162
@support.requires_freebsd_version(11)
@unittest.skipUnless(hasattr(os, 'mkfifo'), 'needs mkfifo()')
def _test_open(self, do_open_close_reader, do_open_close_writer):
filename = support.TESTFN
# Use a fifo: until the child opens it for reading, the parent will
# block when trying to open it for writing.
support.unlink(support.TESTFN)
os.mkfifo(support.TESTFN)
self.addCleanup(support.unlink, support.TESTFN)
support.unlink(filename)
os.mkfifo(filename)
self.addCleanup(support.unlink, filename)
pid = os.fork()
if pid == 0:
# let the parent block
self._sleep()
do_open_close_reader(support.TESTFN)
os._exit(0)
else:
self.addCleanup(os.waitpid, pid, 0)
do_open_close_writer(support.TESTFN)
code = '\n'.join((
'import os, time',
'',
'path = %a' % filename,
'sleep_time = %r' % self.sleep_time,
'',
'# let the parent block',
'time.sleep(sleep_time)',
'',
do_open_close_reader,
))
proc = self.subprocess(code)
with kill_on_error(proc):
do_open_close_writer(filename)
self.assertEqual(proc.wait(), 0)
def python_open(self, path):
fp = open(path, 'w')
fp.close()
def test_open(self):
self._test_open(lambda path: open(path, 'r').close(),
lambda path: open(path, 'w').close())
self._test_open("fp = open(path, 'r')\nfp.close()",
self.python_open)
def os_open(self, path):
fd = os.open(path, os.O_WRONLY)
os.close(fd)
def test_os_open(self):
self._test_open(lambda path: os.close(os.open(path, os.O_RDONLY)),
lambda path: os.close(os.open(path, os.O_WRONLY)))
self._test_open("fd = os.open(path, os.O_RDONLY)\nos.close(fd)",
self.os_open)
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
@ -290,20 +381,22 @@ class SignalEINTRTest(EINTRBaseTest):
old_handler = signal.signal(signum, lambda *args: None)
self.addCleanup(signal.signal, signum, old_handler)
code = '\n'.join((
'import os, time',
'pid = %s' % os.getpid(),
'signum = %s' % int(signum),
'sleep_time = %r' % self.sleep_time,
'time.sleep(sleep_time)',
'os.kill(pid, signum)',
))
t0 = time.monotonic()
child_pid = os.fork()
if child_pid == 0:
# child
try:
self._sleep()
os.kill(pid, signum)
finally:
os._exit(0)
else:
proc = self.subprocess(code)
with kill_on_error(proc):
# parent
signal.sigwaitinfo([signum])
dt = time.monotonic() - t0
os.waitpid(child_pid, 0)
self.assertEqual(proc.wait(), 0)
self.assertGreaterEqual(dt, self.sleep_time)

11
Lib/test/test_eintr.py
View File

@ -1,5 +1,7 @@
import os
import signal
import subprocess
import sys
import unittest
from test import support
@ -14,7 +16,14 @@ class EINTRTests(unittest.TestCase):
# Run the tester in a sub-process, to make sure there is only one
# thread (for reliable signal delivery).
tester = support.findfile("eintr_tester.py", subdir="eintrdata")
script_helper.assert_python_ok(tester)
if support.verbose:
args = [sys.executable, tester]
with subprocess.Popen(args) as proc:
exitcode = proc.wait()
self.assertEqual(exitcode, 0)
else:
script_helper.assert_python_ok(tester)
if __name__ == "__main__":