Skip to content

C
cpython
Kaydet (Commit) eecce795 authored tarafından Georg Brandl's avatar Georg Brandl
Dosyalara gözat
Seçenekler

Convert test_poll to unittest.

üst 3163179f
Hide whitespace changes
Inline Side-by-side
Showing with 140 additions and 195 deletions
Lib/test/output/test_poll deleted 100644 → 0
Dosyayı görüntüle @ 3163179f
test_poll
Running poll test 1
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
This is a test.
Poll test 1 complete
Running poll test 2
Poll test 2 complete
Running poll test 3
Poll test 3 complete
Lib/test/test_poll.py
Dosyayı görüntüle @ eecce795
# Test case for the os.poll() function # Test case for the os.poll() function
import sys, os, select, random import sys, os, select, random, unittest
from test.test_support import verify, verbose, TestSkipped, TESTFN from test.test_support import TestSkipped, TESTFN, run_unittest
try: try:
select.poll select.poll
...@@ -16,177 +16,141 @@ def find_ready_matching(ready, flag): ...@@ -16,177 +16,141 @@ def find_ready_matching(ready, flag):
match.append(fd) match.append(fd)
return match return match
def test_poll1(): class PollTests(unittest.TestCase):
"""Basic functional test of poll object
def test_poll1(self):
Create a bunch of pipe and test that poll works with them. # Basic functional test of poll object
""" # Create a bunch of pipe and test that poll works with them.
print 'Running poll test 1'
p = select.poll() p = select.poll()
NUM_PIPES = 12 NUM_PIPES = 12
MSG = " This is a test." MSG = " This is a test."
MSG_LEN = len(MSG) MSG_LEN = len(MSG)
readers = [] readers = []
writers = [] writers = []
r2w = {} r2w = {}
w2r = {} w2r = {}
for i in range(NUM_PIPES): for i in range(NUM_PIPES):
rd, wr = os.pipe() rd, wr = os.pipe()
p.register(rd, select.POLLIN) p.register(rd, select.POLLIN)
p.register(wr, select.POLLOUT) p.register(wr, select.POLLOUT)
readers.append(rd) readers.append(rd)
writers.append(wr) writers.append(wr)
r2w[rd] = wr r2w[rd] = wr
w2r[wr] = rd w2r[wr] = rd
while writers: bufs = []
ready = p.poll()
ready_writers = find_ready_matching(ready, select.POLLOUT) while writers:
if not ready_writers: ready = p.poll()
raise RuntimeError, "no pipes ready for writing" ready_writers = find_ready_matching(ready, select.POLLOUT)
wr = random.choice(ready_writers) if not ready_writers:
os.write(wr, MSG) raise RuntimeError, "no pipes ready for writing"
wr = random.choice(ready_writers)
ready = p.poll() os.write(wr, MSG)
ready_readers = find_ready_matching(ready, select.POLLIN)
if not ready_readers: ready = p.poll()
raise RuntimeError, "no pipes ready for reading" ready_readers = find_ready_matching(ready, select.POLLIN)
rd = random.choice(ready_readers) if not ready_readers:
buf = os.read(rd, MSG_LEN) raise RuntimeError, "no pipes ready for reading"
verify(len(buf) == MSG_LEN) rd = random.choice(ready_readers)
print buf buf = os.read(rd, MSG_LEN)
os.close(r2w[rd]) ; os.close( rd ) self.assertEqual(len(buf), MSG_LEN)
p.unregister( r2w[rd] ) bufs.append(buf)
p.unregister( rd ) os.close(r2w[rd]) ; os.close( rd )
writers.remove(r2w[rd]) p.unregister( r2w[rd] )
p.unregister( rd )
poll_unit_tests() writers.remove(r2w[rd])
print 'Poll test 1 complete'
self.assertEqual(bufs, [MSG] * NUM_PIPES)
def poll_unit_tests():
# returns NVAL for invalid file descriptor def poll_unit_tests(self):
FD = 42 # returns NVAL for invalid file descriptor
try: FD = 42
os.close(FD) try:
except OSError: os.close(FD)
pass except OSError:
p = select.poll() pass
p.register(FD) p = select.poll()
r = p.poll() p.register(FD)
verify(r[0] == (FD, select.POLLNVAL)) r = p.poll()
self.assertEqual(r[0], (FD, select.POLLNVAL))
f = open(TESTFN, 'w')
fd = f.fileno() f = open(TESTFN, 'w')
p = select.poll() fd = f.fileno()
p.register(f) p = select.poll()
r = p.poll() p.register(f)
verify(r[0][0] == fd) r = p.poll()
f.close() self.assertEqual(r[0][0], fd)
r = p.poll() f.close()
verify(r[0] == (fd, select.POLLNVAL)) r = p.poll()
os.unlink(TESTFN) self.assertEqual(r[0], (fd, select.POLLNVAL))
os.unlink(TESTFN)
# type error for invalid arguments
p = select.poll() # type error for invalid arguments
try: p = select.poll()
p.register(p) self.assertRaises(TypeError, p.register, p)
except TypeError: self.assertRaises(TypeError, p.unregister, p)
pass
else: # can't unregister non-existent object
print "Bogus register call did not raise TypeError" p = select.poll()
try: self.assertRaises(KeyError, p.unregister, 3)
p.unregister(p)
except TypeError: # Test error cases
pass pollster = select.poll()
else: class Nope:
print "Bogus unregister call did not raise TypeError" pass
# can't unregister non-existent object class Almost:
p = select.poll() def fileno(self):
try: return 'fileno'
p.unregister(3)
except KeyError: self.assertRaises(TypeError, pollster.register, Nope(), 0)
pass self.assertRaises(TypeError, pollster.register, Almost(), 0)
else:
print "Bogus unregister call did not raise KeyError" # Another test case for poll(). This is copied from the test case for
# select(), modified to use poll() instead.
# Test error cases
pollster = select.poll() def test_poll2(self):
class Nope: cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
pass p = os.popen(cmd, 'r')
pollster = select.poll()
class Almost: pollster.register( p, select.POLLIN )
def fileno(self): for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10:
return 'fileno' fdlist = pollster.poll(tout)
if (fdlist == []):
try: continue
pollster.register( Nope(), 0 ) fd, flags = fdlist[0]
except TypeError: pass if flags & select.POLLHUP:
else: print 'expected TypeError exception, not raised' line = p.readline()
if line != "":
try: self.fail('error: pipe seems to be closed, but still returns data')
pollster.register( Almost(), 0 ) continue
except TypeError: pass
else: print 'expected TypeError exception, not raised' elif flags & select.POLLIN:
line = p.readline()
if not line:
# Another test case for poll(). This is copied from the test case for break
# select(), modified to use poll() instead. continue
else:
def test_poll2(): self.fail('Unexpected return value from select.poll: %s' % fdlist)
print 'Running poll test 2' p.close()
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') def test_poll3(self):
pollster = select.poll() # test int overflow
pollster.register( p, select.POLLIN ) pollster = select.poll()
for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10: pollster.register(1)
if verbose:
print 'timeout =', tout self.assertRaises(OverflowError, pollster.poll, 1L << 64)
fdlist = pollster.poll(tout)
if (fdlist == []): x = 2 + 3
continue if x != 5:
fd, flags = fdlist[0] self.fail('Overflow must have occurred')
if flags & select.POLLHUP:
line = p.readline() def test_main():
if line != "": run_unittest(PollTests)
print 'error: pipe seems to be closed, but still returns data'
continue if __name__ == '__main__':
test_main()
elif flags & select.POLLIN:
line = p.readline()
if verbose:
print repr(line)
if not line:
if verbose:
print 'EOF'
break
continue
else:
print 'Unexpected return value from select.poll:', fdlist
p.close()
print 'Poll test 2 complete'
def test_poll3():
# test int overflow
print 'Running poll test 3'
pollster = select.poll()
pollster.register(1)
try:
pollster.poll(1L << 64)
except OverflowError:
pass
else:
print 'Expected OverflowError with excessive timeout'
x = 2 + 3
if x != 5:
print 'Overflow must have occurred'
print 'Poll test 3 complete'
test_poll1()
test_poll2()
test_poll3()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment