[Python-3000-checkins] r66024 - in python/branches/py3k: Lib/multiprocessing/connection.py Lib/test/test_multiprocessing.py

[neal.norwitz]

Author: neal.norwitz
Date: Mon Aug 25 03:53:32 2008
New Revision: 66024

Log:
Merged revisions 66023 via svnmerge from
svn+ssh://pythondev at 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
........


Modified:
python/branches/py3k/ (props changed)
python/branches/py3k/Lib/multiprocessing/connection.py
python/branches/py3k/Lib/test/test_multiprocessing.py

Modified: python/branches/py3k/Lib/multiprocessing/connection.py
==============================================================================
--- python/branches/py3k/Lib/multiprocessing/connection.py (original)
+++ python/branches/py3k/Lib/multiprocessing/connection.py Mon Aug 25 03:53:32 2008
@@ -17,7 +17,7 @@
import itertools

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.forking import duplicate, close


Modified: python/branches/py3k/Lib/test/test_multiprocessing.py
==============================================================================
--- python/branches/py3k/Lib/test/test_multiprocessing.py (original)
+++ python/branches/py3k/Lib/test/test_multiprocessing.py Mon Aug 25 03:53:32 2008
@@ -1737,6 +1737,37 @@
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 @@
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