mirror of
https://github.com/python/cpython.git
synced 2025-09-16 21:56:14 +00:00
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
971f10210e
commit
0c519b3a5e
2 changed files with 34 additions and 2 deletions
|
@ -17,7 +17,7 @@ import tempfile
|
||||||
import itertools
|
import itertools
|
||||||
|
|
||||||
import _multiprocessing
|
import _multiprocessing
|
||||||
from multiprocessing import current_process
|
from multiprocessing import current_process, AuthenticationError
|
||||||
from multiprocessing.util import get_temp_dir, Finalize, sub_debug, debug
|
from multiprocessing.util import get_temp_dir, Finalize, sub_debug, debug
|
||||||
from multiprocessing.forking import duplicate, close
|
from multiprocessing.forking import duplicate, close
|
||||||
|
|
||||||
|
|
|
@ -1736,6 +1736,37 @@ class ThreadsMixin(object):
|
||||||
testcases_threads = create_test_cases(ThreadsMixin, type='threads')
|
testcases_threads = create_test_cases(ThreadsMixin, type='threads')
|
||||||
globals().update(testcases_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]
|
||||||
|
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
|
@ -1764,7 +1795,8 @@ def test_main(run=None):
|
||||||
testcases = (
|
testcases = (
|
||||||
sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
|
sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
|
||||||
sorted(testcases_threads.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
|
loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue