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
........

diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py
index b962060615bd31eb3927d418e2c2c8d315b15f52..1c4d48e5cf23d12dad18ac7175cf98948c8bb39e 100644 (file)
--- a/Lib/multiprocessing/connection.py
+++ b/Lib/multiprocessing/connection.py
@@ -17,7 +17,7 @@ import tempfile
 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
 

diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py
index 102042d9b60d85cf3d1153fee5429306c6abadf0..891795de6c84ee0d64fd736a349c60a3d856a1e4 100644 (file)
--- a/Lib/test/test_multiprocessing.py
+++ b/Lib/test/test_multiprocessing.py
@@ -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