Selectively enable tests for asyncore.readwrite based on the presence
of poll support in the select module (since this is the only case in which readwrite can be called). [GSoC - Alan McIntyre]
This commit is contained in:
parent
6819210b9e
commit
7f4f41255f
|
@ -106,87 +106,83 @@ class HelperFunctionTests(unittest.TestCase):
|
|||
asyncore._exception(tr2)
|
||||
self.assertEqual(tr2.error_handled, True)
|
||||
|
||||
## Commented out these tests because test a non-documented function
|
||||
## (which is actually public, why it's not documented?). Anyway, the
|
||||
## tests *and* the function uses constants in the select module that
|
||||
## are not present in Windows systems (see this thread:
|
||||
## http://mail.python.org/pipermail/python-list/2001-October/109973.html)
|
||||
## Note even that these constants are mentioned in the select
|
||||
## documentation, as a parameter of "poll" method "register", but are
|
||||
## not explicit declared as constants of the module.
|
||||
## . Facundo Batista
|
||||
##
|
||||
## def test_readwrite(self):
|
||||
## # Check that correct methods are called by readwrite()
|
||||
##
|
||||
## class testobj:
|
||||
## def __init__(self):
|
||||
## self.read = False
|
||||
## self.write = False
|
||||
## self.expt = False
|
||||
##
|
||||
## def handle_read_event(self):
|
||||
## self.read = True
|
||||
##
|
||||
## def handle_write_event(self):
|
||||
## self.write = True
|
||||
##
|
||||
## def handle_expt_event(self):
|
||||
## self.expt = True
|
||||
##
|
||||
## def handle_error(self):
|
||||
## self.error_handled = True
|
||||
##
|
||||
## for flag in (select.POLLIN, select.POLLPRI):
|
||||
## tobj = testobj()
|
||||
## self.assertEqual(tobj.read, False)
|
||||
## asyncore.readwrite(tobj, flag)
|
||||
## self.assertEqual(tobj.read, True)
|
||||
##
|
||||
## # check that ExitNow exceptions in the object handler method
|
||||
## # bubbles all the way up through asyncore readwrite call
|
||||
## tr1 = exitingdummy()
|
||||
## self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
|
||||
##
|
||||
## # check that an exception other than ExitNow in the object handler
|
||||
## # method causes the handle_error method to get called
|
||||
## tr2 = crashingdummy()
|
||||
## asyncore.readwrite(tr2, flag)
|
||||
## self.assertEqual(tr2.error_handled, True)
|
||||
##
|
||||
## tobj = testobj()
|
||||
## self.assertEqual(tobj.write, False)
|
||||
## asyncore.readwrite(tobj, select.POLLOUT)
|
||||
## self.assertEqual(tobj.write, True)
|
||||
##
|
||||
## # check that ExitNow exceptions in the object handler method
|
||||
## # bubbles all the way up through asyncore readwrite call
|
||||
## tr1 = exitingdummy()
|
||||
## self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1,
|
||||
## select.POLLOUT)
|
||||
##
|
||||
## # check that an exception other than ExitNow in the object handler
|
||||
## # method causes the handle_error method to get called
|
||||
## tr2 = crashingdummy()
|
||||
## asyncore.readwrite(tr2, select.POLLOUT)
|
||||
## self.assertEqual(tr2.error_handled, True)
|
||||
##
|
||||
## for flag in (select.POLLERR, select.POLLHUP, select.POLLNVAL):
|
||||
## tobj = testobj()
|
||||
## self.assertEqual(tobj.expt, False)
|
||||
## asyncore.readwrite(tobj, flag)
|
||||
## self.assertEqual(tobj.expt, True)
|
||||
##
|
||||
## # check that ExitNow exceptions in the object handler method
|
||||
## # bubbles all the way up through asyncore readwrite calls
|
||||
## tr1 = exitingdummy()
|
||||
## self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
|
||||
##
|
||||
## # check that an exception other than ExitNow in the object handler
|
||||
## # method causes the handle_error method to get called
|
||||
## tr2 = crashingdummy()
|
||||
## asyncore.readwrite(tr2, flag)
|
||||
## self.assertEqual(tr2.error_handled, True)
|
||||
# asyncore.readwrite uses constants in the select module that
|
||||
# are not present in Windows systems (see this thread:
|
||||
# http://mail.python.org/pipermail/python-list/2001-October/109973.html)
|
||||
# These constants should be present as long as poll is available
|
||||
|
||||
if hasattr(select, 'poll'):
|
||||
def test_readwrite(self):
|
||||
# Check that correct methods are called by readwrite()
|
||||
|
||||
class testobj:
|
||||
def __init__(self):
|
||||
self.read = False
|
||||
self.write = False
|
||||
self.expt = False
|
||||
|
||||
def handle_read_event(self):
|
||||
self.read = True
|
||||
|
||||
def handle_write_event(self):
|
||||
self.write = True
|
||||
|
||||
def handle_expt_event(self):
|
||||
self.expt = True
|
||||
|
||||
def handle_error(self):
|
||||
self.error_handled = True
|
||||
|
||||
for flag in (select.POLLIN, select.POLLPRI):
|
||||
tobj = testobj()
|
||||
self.assertEqual(tobj.read, False)
|
||||
asyncore.readwrite(tobj, flag)
|
||||
self.assertEqual(tobj.read, True)
|
||||
|
||||
# check that ExitNow exceptions in the object handler method
|
||||
# bubbles all the way up through asyncore readwrite call
|
||||
tr1 = exitingdummy()
|
||||
self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
|
||||
|
||||
# check that an exception other than ExitNow in the object handler
|
||||
# method causes the handle_error method to get called
|
||||
tr2 = crashingdummy()
|
||||
asyncore.readwrite(tr2, flag)
|
||||
self.assertEqual(tr2.error_handled, True)
|
||||
|
||||
tobj = testobj()
|
||||
self.assertEqual(tobj.write, False)
|
||||
asyncore.readwrite(tobj, select.POLLOUT)
|
||||
self.assertEqual(tobj.write, True)
|
||||
|
||||
# check that ExitNow exceptions in the object handler method
|
||||
# bubbles all the way up through asyncore readwrite call
|
||||
tr1 = exitingdummy()
|
||||
self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1,
|
||||
select.POLLOUT)
|
||||
|
||||
# check that an exception other than ExitNow in the object handler
|
||||
# method causes the handle_error method to get called
|
||||
tr2 = crashingdummy()
|
||||
asyncore.readwrite(tr2, select.POLLOUT)
|
||||
self.assertEqual(tr2.error_handled, True)
|
||||
|
||||
for flag in (select.POLLERR, select.POLLHUP, select.POLLNVAL):
|
||||
tobj = testobj()
|
||||
self.assertEqual(tobj.expt, False)
|
||||
asyncore.readwrite(tobj, flag)
|
||||
self.assertEqual(tobj.expt, True)
|
||||
|
||||
# check that ExitNow exceptions in the object handler method
|
||||
# bubbles all the way up through asyncore readwrite calls
|
||||
tr1 = exitingdummy()
|
||||
self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
|
||||
|
||||
# check that an exception other than ExitNow in the object handler
|
||||
# method causes the handle_error method to get called
|
||||
tr2 = crashingdummy()
|
||||
asyncore.readwrite(tr2, flag)
|
||||
self.assertEqual(tr2.error_handled, True)
|
||||
|
||||
def test_closeall(self):
|
||||
self.closeall_check(False)
|
||||
|
|
Loading…
Reference in New Issue