diff options
author | Ezio Melotti <ezio.melotti@gmail.com> | 2010-07-27 23:45:05 (GMT) |
---|---|---|
committer | Ezio Melotti <ezio.melotti@gmail.com> | 2010-07-27 23:45:05 (GMT) |
commit | 3ff4463b356b79109fd7f71ef4507fb079871659 (patch) | |
tree | 2a8b320ad997da36a9c2c57c2c53536eb1dc289b | |
parent | e8d2eee0fada6b6a7c90c260366d30c0b0c16ab2 (diff) | |
download | cpython-3ff4463b356b79109fd7f71ef4507fb079871659.zip cpython-3ff4463b356b79109fd7f71ef4507fb079871659.tar.gz cpython-3ff4463b356b79109fd7f71ef4507fb079871659.tar.bz2 |
Merged revisions 83183,83186 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/branches/py3k
........
r83183 | ezio.melotti | 2010-07-28 01:03:33 +0300 (Wed, 28 Jul 2010) | 1 line
Use proper skips and assert* methods in test_asyncore.
........
r83186 | ezio.melotti | 2010-07-28 01:24:13 +0300 (Wed, 28 Jul 2010) | 1 line
With skipUnless there is no need to add test classes conditionally.
........
-rw-r--r-- | Lib/test/test_asyncore.py | 193 |
1 files changed, 96 insertions, 97 deletions
diff --git a/Lib/test/test_asyncore.py b/Lib/test/test_asyncore.py index 6973f4b..c352c39 100644 --- a/Lib/test/test_asyncore.py +++ b/Lib/test/test_asyncore.py @@ -117,65 +117,65 @@ class HelperFunctionTests(unittest.TestCase): # 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() - - attributes = ('read', 'expt', 'write', 'closed', 'error_handled') - - expected = ( - (select.POLLIN, 'read'), - (select.POLLPRI, 'expt'), - (select.POLLOUT, 'write'), - (select.POLLERR, 'closed'), - (select.POLLHUP, 'closed'), - (select.POLLNVAL, 'closed'), - ) - - class testobj: - def __init__(self): - self.read = False - self.write = False - self.closed = False - self.expt = False - self.error_handled = False - - def handle_read_event(self): - self.read = True - - def handle_write_event(self): - self.write = True - - def handle_close(self): - self.closed = True - - def handle_expt_event(self): - self.expt = True - - def handle_error(self): - self.error_handled = True - - for flag, expectedattr in expected: - tobj = testobj() - self.assertEqual(getattr(tobj, expectedattr), False) - asyncore.readwrite(tobj, flag) - - # Only the attribute modified by the routine we expect to be - # called should be True. - for attr in attributes: - self.assertEqual(getattr(tobj, attr), attr==expectedattr) - - # 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() - self.assertEqual(tr2.error_handled, False) - asyncore.readwrite(tr2, flag) - self.assertEqual(tr2.error_handled, True) + @unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required') + def test_readwrite(self): + # Check that correct methods are called by readwrite() + + attributes = ('read', 'expt', 'write', 'closed', 'error_handled') + + expected = ( + (select.POLLIN, 'read'), + (select.POLLPRI, 'expt'), + (select.POLLOUT, 'write'), + (select.POLLERR, 'closed'), + (select.POLLHUP, 'closed'), + (select.POLLNVAL, 'closed'), + ) + + class testobj: + def __init__(self): + self.read = False + self.write = False + self.closed = False + self.expt = False + self.error_handled = False + + def handle_read_event(self): + self.read = True + + def handle_write_event(self): + self.write = True + + def handle_close(self): + self.closed = True + + def handle_expt_event(self): + self.expt = True + + def handle_error(self): + self.error_handled = True + + for flag, expectedattr in expected: + tobj = testobj() + self.assertEqual(getattr(tobj, expectedattr), False) + asyncore.readwrite(tobj, flag) + + # Only the attribute modified by the routine we expect to be + # called should be True. + for attr in attributes: + self.assertEqual(getattr(tobj, attr), attr==expectedattr) + + # 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() + self.assertEqual(tr2.error_handled, False) + asyncore.readwrite(tr2, flag) + self.assertEqual(tr2.error_handled, True) def test_closeall(self): self.closeall_check(False) @@ -258,7 +258,7 @@ class DispatcherTests(unittest.TestCase): sys.stderr = stderr lines = fp.getvalue().splitlines() - self.assertEquals(lines, ['log: %s' % l1, 'log: %s' % l2]) + self.assertEqual(lines, ['log: %s' % l1, 'log: %s' % l2]) def test_log_info(self): d = asyncore.dispatcher() @@ -280,7 +280,7 @@ class DispatcherTests(unittest.TestCase): lines = fp.getvalue().splitlines() expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3] - self.assertEquals(lines, expected) + self.assertEqual(lines, expected) def test_unhandled(self): d = asyncore.dispatcher() @@ -305,7 +305,7 @@ class DispatcherTests(unittest.TestCase): 'warning: unhandled write event', 'warning: unhandled connect event', 'warning: unhandled accept event'] - self.assertEquals(lines, expected) + self.assertEqual(lines, expected) def test_issue_8594(self): # XXX - this test is supposed to be removed in next major Python @@ -321,7 +321,7 @@ class DispatcherTests(unittest.TestCase): warnings.simplefilter("always") family = d.family self.assertEqual(family, socket.AF_INET) - self.assertTrue(len(w) == 1) + self.assertEqual(len(w), 1) self.assertTrue(issubclass(w[0].category, DeprecationWarning)) def test_strerror(self): @@ -330,7 +330,7 @@ class DispatcherTests(unittest.TestCase): if hasattr(os, 'strerror'): self.assertEqual(err, os.strerror(errno.EPERM)) err = asyncore._strerror(-1) - self.assertTrue("unknown error" in err.lower()) + self.assertIn("unknown error", err.lower()) class dispatcherwithsend_noread(asyncore.dispatcher_with_send): @@ -393,38 +393,40 @@ class DispatcherWithSendTests(unittest.TestCase): class DispatcherWithSendTests_UsePoll(DispatcherWithSendTests): usepoll = True -if hasattr(asyncore, 'file_wrapper'): - class FileWrapperTest(unittest.TestCase): - def setUp(self): - self.d = "It's not dead, it's sleeping!" - file(TESTFN, 'w').write(self.d) +@unittest.skipUnless(hasattr(asyncore, 'file_wrapper'), + 'asyncore.file_wrapper required') +class FileWrapperTest(unittest.TestCase): + def setUp(self): + self.d = "It's not dead, it's sleeping!" + file(TESTFN, 'w').write(self.d) + + def tearDown(self): + unlink(TESTFN) - def tearDown(self): - unlink(TESTFN) + def test_recv(self): + fd = os.open(TESTFN, os.O_RDONLY) + w = asyncore.file_wrapper(fd) + os.close(fd) - def test_recv(self): - fd = os.open(TESTFN, os.O_RDONLY) - w = asyncore.file_wrapper(fd) - os.close(fd) + self.assertNotEqual(w.fd, fd) + self.assertNotEqual(w.fileno(), fd) + self.assertEqual(w.recv(13), "It's not dead") + self.assertEqual(w.read(6), ", it's") + w.close() + self.assertRaises(OSError, w.read, 1) - self.assertNotEqual(w.fd, fd) - self.assertNotEqual(w.fileno(), fd) - self.assertEqual(w.recv(13), "It's not dead") - self.assertEqual(w.read(6), ", it's") - w.close() - self.assertRaises(OSError, w.read, 1) - def test_send(self): - d1 = "Come again?" - d2 = "I want to buy some cheese." - fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND) - w = asyncore.file_wrapper(fd) - os.close(fd) + def test_send(self): + d1 = "Come again?" + d2 = "I want to buy some cheese." + fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND) + w = asyncore.file_wrapper(fd) + os.close(fd) - w.write(d1) - w.send(d2) - w.close() - self.assertEqual(file(TESTFN).read(), self.d + d1 + d2) + w.write(d1) + w.send(d2) + w.close() + self.assertEqual(file(TESTFN).read(), self.d + d1 + d2) class BaseTestHandler(asyncore.dispatcher): @@ -691,18 +693,15 @@ class BaseTestAPI(unittest.TestCase): class TestAPI_UseSelect(BaseTestAPI): use_poll = False +@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required') class TestAPI_UsePoll(BaseTestAPI): use_poll = True def test_main(): tests = [HelperFunctionTests, DispatcherTests, DispatcherWithSendTests, - DispatcherWithSendTests_UsePoll, TestAPI_UseSelect] - if hasattr(asyncore, 'file_wrapper'): - tests.append(FileWrapperTest) - if hasattr(select, 'poll'): - tests.append(TestAPI_UsePoll) - + DispatcherWithSendTests_UsePoll, TestAPI_UseSelect, + TestAPI_UsePoll, FileWrapperTest] run_unittest(*tests) if __name__ == "__main__": |