mirror of
https://github.com/python/cpython.git
synced 2025-08-04 17:08:35 +00:00
Merged revisions 66023 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk ........ r66023 | neal.norwitz | 2008-08-24 18:50:24 -0700 (Sun, 24 Aug 2008) | 6 lines Fix problem reported by pychecker where AuthenticationError wasn't imported. Add some test coverage to this code. More tests should be added (TODO added). R=Brett TESTED=./python -E -tt ./Lib/test/regrtest.py test_multiprocessing ........
This commit is contained in:
parent
c34177ccb2
commit
5d6415ee26
2 changed files with 34 additions and 2 deletions
|
@ -1737,6 +1737,37 @@ class ThreadsMixin(object):
|
|||
testcases_threads = create_test_cases(ThreadsMixin, type='threads')
|
||||
globals().update(testcases_threads)
|
||||
|
||||
class OtherTest(unittest.TestCase):
|
||||
# TODO: add more tests for deliver/answer challenge.
|
||||
def test_deliver_challenge_auth_failure(self):
|
||||
class _FakeConnection(object):
|
||||
def recv_bytes(self, size):
|
||||
return 'something bogus'
|
||||
def send_bytes(self, data):
|
||||
pass
|
||||
self.assertRaises(multiprocessing.AuthenticationError,
|
||||
multiprocessing.connection.deliver_challenge,
|
||||
_FakeConnection(), b'abc')
|
||||
|
||||
def test_answer_challenge_auth_failure(self):
|
||||
class _FakeConnection(object):
|
||||
def __init__(self):
|
||||
self.count = 0
|
||||
def recv_bytes(self, size):
|
||||
self.count += 1
|
||||
if self.count == 1:
|
||||
return multiprocessing.connection.CHALLENGE
|
||||
elif self.count == 2:
|
||||
return 'something bogus'
|
||||
return ''
|
||||
def send_bytes(self, data):
|
||||
pass
|
||||
self.assertRaises(multiprocessing.AuthenticationError,
|
||||
multiprocessing.connection.answer_challenge,
|
||||
_FakeConnection(), b'abc')
|
||||
|
||||
testcases_other = [OtherTest]
|
||||
|
||||
#
|
||||
#
|
||||
#
|
||||
|
@ -1765,7 +1796,8 @@ def test_main(run=None):
|
|||
testcases = (
|
||||
sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
|
||||
sorted(testcases_threads.values(), key=lambda tc:tc.__name__) +
|
||||
sorted(testcases_manager.values(), key=lambda tc:tc.__name__)
|
||||
sorted(testcases_manager.values(), key=lambda tc:tc.__name__) +
|
||||
testcases_other
|
||||
)
|
||||
|
||||
loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue