mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-04 03:44:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

Changed the used port and commented out some tests that uses

Browse source 
a non documented function that appers to uses resources
not present in Windows.
This commit is contained in:
Facundo Batista 2007-07-15 20:30:39 +00:00
parent f47bc182f1
commit 66263cf151
1 changed files with 82 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

154
Lib/test/test_asyncore.py
View file

@ -12,7 +12,7 @@ from test.test_support import TESTFN, run_unittest, unlink
from StringIO import StringIO from StringIO import StringIO
HOST = "127.0.0.1" HOST = "127.0.0.1"
PORT = 54322 PORT = 54329
class dummysocket: class dummysocket:
def __init__(self): def __init__(self):
@ -96,77 +96,87 @@ class HelperFunctionTests(unittest.TestCase):
asyncore._exception(tr2) asyncore._exception(tr2)
self.assertEqual(tr2.error_handled, True) self.assertEqual(tr2.error_handled, True)
def test_readwrite(self): ## Commented out these tests because test a non-documented function
# Check that correct methods are called by readwrite() ## (which is actually public, why it's not documented?). Anyway, the
## tests *and* the function uses constants in the select module that
class testobj: ## are not present in Windows systems (see this thread:
def __init__(self): ## http://mail.python.org/pipermail/python-list/2001-October/109973.html)
self.read = False ## Note even that these constants are mentioned in the select
self.write = False ## documentation, as a parameter of "poll" method "register", but are
self.expt = False ## not explicit declared as constants of the module.
## . Facundo Batista
def handle_read_event(self): ##
self.read = True ## def test_readwrite(self):
## # Check that correct methods are called by readwrite()
def handle_write_event(self): ##
self.write = True ## class testobj:
## def __init__(self):
def handle_expt_event(self): ## self.read = False
self.expt = True ## self.write = False
## self.expt = False
def handle_error(self): ##
self.error_handled = True ## def handle_read_event(self):
## self.read = True
for flag in (select.POLLIN, select.POLLPRI): ##
tobj = testobj() ## def handle_write_event(self):
self.assertEqual(tobj.read, False) ## self.write = True
asyncore.readwrite(tobj, flag) ##
self.assertEqual(tobj.read, True) ## def handle_expt_event(self):
## self.expt = True
# check that ExitNow exceptions in the object handler method ##
# bubbles all the way up through asyncore readwrite call ## def handle_error(self):
tr1 = exitingdummy() ## self.error_handled = True
self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag) ##
## for flag in (select.POLLIN, select.POLLPRI):
# check that an exception other than ExitNow in the object handler ## tobj = testobj()
# method causes the handle_error method to get called ## self.assertEqual(tobj.read, False)
tr2 = crashingdummy() ## asyncore.readwrite(tobj, flag)
asyncore.readwrite(tr2, flag) ## self.assertEqual(tobj.read, True)
self.assertEqual(tr2.error_handled, True) ##
## # check that ExitNow exceptions in the object handler method
tobj = testobj() ## # bubbles all the way up through asyncore readwrite call
self.assertEqual(tobj.write, False) ## tr1 = exitingdummy()
asyncore.readwrite(tobj, select.POLLOUT) ## self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
self.assertEqual(tobj.write, True) ##
## # check that an exception other than ExitNow in the object handler
# check that ExitNow exceptions in the object handler method ## # method causes the handle_error method to get called
# bubbles all the way up through asyncore readwrite call ## tr2 = crashingdummy()
tr1 = exitingdummy() ## asyncore.readwrite(tr2, flag)
self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, ## self.assertEqual(tr2.error_handled, True)
select.POLLOUT) ##
## tobj = testobj()
# check that an exception other than ExitNow in the object handler ## self.assertEqual(tobj.write, False)
# method causes the handle_error method to get called ## asyncore.readwrite(tobj, select.POLLOUT)
tr2 = crashingdummy() ## self.assertEqual(tobj.write, True)
asyncore.readwrite(tr2, select.POLLOUT) ##
self.assertEqual(tr2.error_handled, True) ## # check that ExitNow exceptions in the object handler method
## # bubbles all the way up through asyncore readwrite call
for flag in (select.POLLERR, select.POLLHUP, select.POLLNVAL): ## tr1 = exitingdummy()
tobj = testobj() ## self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1,
self.assertEqual(tobj.expt, False) ## select.POLLOUT)
asyncore.readwrite(tobj, flag) ##
self.assertEqual(tobj.expt, True) ## # check that an exception other than ExitNow in the object handler
## # method causes the handle_error method to get called
# check that ExitNow exceptions in the object handler method ## tr2 = crashingdummy()
# bubbles all the way up through asyncore readwrite calls ## asyncore.readwrite(tr2, select.POLLOUT)
tr1 = exitingdummy() ## self.assertEqual(tr2.error_handled, True)
self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag) ##
## for flag in (select.POLLERR, select.POLLHUP, select.POLLNVAL):
# check that an exception other than ExitNow in the object handler ## tobj = testobj()
# method causes the handle_error method to get called ## self.assertEqual(tobj.expt, False)
tr2 = crashingdummy() ## asyncore.readwrite(tobj, flag)
asyncore.readwrite(tr2, flag) ## self.assertEqual(tobj.expt, True)
self.assertEqual(tr2.error_handled, 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): def test_closeall(self):
self.closeall_check(False) self.closeall_check(False)