summaryrefslogtreecommitdiff
path: root/lib/python2.7/test/test_poll.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/python2.7/test/test_poll.py')
-rw-r--r--lib/python2.7/test/test_poll.py167
1 files changed, 0 insertions, 167 deletions
diff --git a/lib/python2.7/test/test_poll.py b/lib/python2.7/test/test_poll.py
deleted file mode 100644
index 55294f8..0000000
--- a/lib/python2.7/test/test_poll.py
+++ /dev/null
@@ -1,167 +0,0 @@
-# Test case for the os.poll() function
-
-import os, select, random, unittest
-import _testcapi
-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')
-
- pollster = select.poll()
- # Issue 15989
- self.assertRaises(OverflowError, pollster.register, 0,
- _testcapi.SHRT_MAX + 1)
- self.assertRaises(OverflowError, pollster.register, 0,
- _testcapi.USHRT_MAX + 1)
- self.assertRaises(OverflowError, pollster.poll, _testcapi.INT_MAX + 1)
- self.assertRaises(OverflowError, pollster.poll, _testcapi.UINT_MAX + 1)
-
-def test_main():
- run_unittest(PollTests)
-
-if __name__ == '__main__':
- test_main()