Message137057
+ self.assertRaises(TypeError, os.pipe2, (0, 0))
Do you want to call the function with two arguments or one tuple with 2 items? You may test both :-)
+ # try a write big enough to fill-up the pipe (64K on most kernels): this
+ # should perform a partial write, not block
+ os.write(w, b'x' * 100000)
This constant should be moved to test.support. BaseTestCase.test_communicate_pipe_buf() on subprocess uses:
x, y = os.pipe()
if mswindows:
pipe_buf = 512
else:
pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")
SignalsTest.check_interrupted_write() of test_io uses:
# 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))
Instead of a constant, it may be function taking a pipe end as argument and returning the size of its buffer. Something like:
def pipe_buffer_size(fd):
if hasattr(os, 'fpathconf'): # better than sys.platform == "win32"?
pipe_buf = 512
else:
pipe_buf = os.fpathconf(fd, "PC_PIPE_BUF")
+ self.assertTrue((time.time() - start) < 1.0, "Test took too long for O_NONBLOCK.")
Hum, I'm not sure that it's revelant to test the time: if the call blocks, it will block forever. If the system is busy, the read+write may takes longer than 1 second. |
|
Date |
User |
Action |
Args |
2011-05-27 14:34:37 | vstinner | set | recipients:
+ vstinner, pitrou, neologix |
2011-05-27 14:34:37 | vstinner | set | messageid: <1306506877.39.0.678592851218.issue12196@psf.upfronthosting.co.za> |
2011-05-27 14:34:36 | vstinner | link | issue12196 messages |
2011-05-27 14:34:36 | vstinner | create | |
|