# Test case for the os.poll() function | |
import os, select, random, unittest | |
from test.test_support import TESTFN, run_unittest | |
try: | |
select.poll | |
except AttributeError: | |
raise unittest.SkipTest, "select.poll not defined -- skipping test_poll" | |
def find_ready_matching(ready, flag): | |
match = [] | |
for fd, mode in ready: | |
if mode & flag: | |
match.append(fd) | |
return match | |
class PollTests(unittest.TestCase): | |
def test_poll1(self): | |
# Basic functional test of poll object | |
# Create a bunch of pipe and test that poll works with them. | |
p = select.poll() | |
NUM_PIPES = 12 | |
MSG = " This is a test." | |
MSG_LEN = len(MSG) | |
readers = [] | |
writers = [] | |
r2w = {} | |
w2r = {} | |
for i in range(NUM_PIPES): | |
rd, wr = os.pipe() | |
p.register(rd) | |
p.modify(rd, select.POLLIN) | |
p.register(wr, select.POLLOUT) | |
readers.append(rd) | |
writers.append(wr) | |
r2w[rd] = wr | |
w2r[wr] = rd | |
bufs = [] | |
while writers: | |
ready = p.poll() | |
ready_writers = find_ready_matching(ready, select.POLLOUT) | |
if not ready_writers: | |
raise RuntimeError, "no pipes ready for writing" | |
wr = random.choice(ready_writers) | |
os.write(wr, MSG) | |
ready = p.poll() | |
ready_readers = find_ready_matching(ready, select.POLLIN) | |
if not ready_readers: | |
raise RuntimeError, "no pipes ready for reading" | |
rd = random.choice(ready_readers) | |
buf = os.read(rd, MSG_LEN) | |
self.assertEqual(len(buf), MSG_LEN) | |
bufs.append(buf) | |
os.close(r2w[rd]) ; os.close( rd ) | |
p.unregister( r2w[rd] ) | |
p.unregister( rd ) | |
writers.remove(r2w[rd]) | |
self.assertEqual(bufs, [MSG] * NUM_PIPES) | |
def poll_unit_tests(self): | |
# returns NVAL for invalid file descriptor | |
FD = 42 | |
try: | |
os.close(FD) | |
except OSError: | |
pass | |
p = select.poll() | |
p.register(FD) | |
r = p.poll() | |
self.assertEqual(r[0], (FD, select.POLLNVAL)) | |
f = open(TESTFN, 'w') | |
fd = f.fileno() | |
p = select.poll() | |
p.register(f) | |
r = p.poll() | |
self.assertEqual(r[0][0], fd) | |
f.close() | |
r = p.poll() | |
self.assertEqual(r[0], (fd, select.POLLNVAL)) | |
os.unlink(TESTFN) | |
# type error for invalid arguments | |
p = select.poll() | |
self.assertRaises(TypeError, p.register, p) | |
self.assertRaises(TypeError, p.unregister, p) | |
# can't unregister non-existent object | |
p = select.poll() | |
self.assertRaises(KeyError, p.unregister, 3) | |
# Test error cases | |
pollster = select.poll() | |
class Nope: | |
pass | |
class Almost: | |
def fileno(self): | |
return 'fileno' | |
self.assertRaises(TypeError, pollster.register, Nope(), 0) | |
self.assertRaises(TypeError, pollster.register, Almost(), 0) | |
# Another test case for poll(). This is copied from the test case for | |
# select(), modified to use poll() instead. | |
def test_poll2(self): | |
cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done' | |
p = os.popen(cmd, 'r') | |
pollster = select.poll() | |
pollster.register( p, select.POLLIN ) | |
for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10: | |
fdlist = pollster.poll(tout) | |
if (fdlist == []): | |
continue | |
fd, flags = fdlist[0] | |
if flags & select.POLLHUP: | |
line = p.readline() | |
if line != "": | |
self.fail('error: pipe seems to be closed, but still returns data') | |
continue | |
elif flags & select.POLLIN: | |
line = p.readline() | |
if not line: | |
break | |
continue | |
else: | |
self.fail('Unexpected return value from select.poll: %s' % fdlist) | |
p.close() | |
def test_poll3(self): | |
# test int overflow | |
pollster = select.poll() | |
pollster.register(1) | |
self.assertRaises(OverflowError, pollster.poll, 1L << 64) | |
x = 2 + 3 | |
if x != 5: | |
self.fail('Overflow must have occurred') | |
def test_main(): | |
run_unittest(PollTests) | |
if __name__ == '__main__': | |
test_main() |