[Python-checkins] python/dist/src/Lib/test test_socket_ssl.py,1.6,1.7

[nnorwitz@users.sourceforge.net]

Update of /cvsroot/python/python/dist/src/Lib/test
In directory sc8-pr-cvs1:/tmp/cvs-serv13018/Lib/test

Modified Files:
	test_socket_ssl.py 
Log Message:
Fix SF #754870, SSL crash interpreter when remote side closes during connect

Also fix a memory leak.


Index: test_socket_ssl.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/test/test_socket_ssl.py,v
retrieving revision 1.6
retrieving revision 1.7
diff -C2 -d -r1.6 -r1.7
*** test_socket_ssl.py	4 Dec 2002 03:26:57 -0000	1.6
--- test_socket_ssl.py	30 Jun 2003 03:25:20 -0000	1.7
***************
*** 3,6 ****
--- 3,7 ----
  from test import test_support
  import socket
+ import time
  
  # Optionally test SSL support.  This requires the 'network' resource as given
***************
*** 9,13 ****
                       hasattr(socket, "ssl"))
  
! def test_main():
      test_support.requires('network')
      if not hasattr(socket, "ssl"):
--- 10,14 ----
                       hasattr(socket, "ssl"))
  
! def test_basic():
      test_support.requires('network')
      if not hasattr(socket, "ssl"):
***************
*** 28,31 ****
--- 29,67 ----
      buf = f.read()
      f.close()
+ 
+ def test_rude_shutdown():
+     try:
+         import thread
+     except ImportError:
+         return
+ 
+     # some random port to connect to
+     PORT = 9934
+     def listener():
+         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+         s.bind(('', PORT))
+         s.listen(5)
+         s.accept()
+         del s
+         thread.exit()
+ 
+     def connector():
+         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+         s.connect(('', PORT))
+         try:
+             ssl_sock = socket.ssl(s)
+         except socket.sslerror:
+             pass
+         else:
+             raise test_support.TestFailed, \
+                         'connecting to closed SSL socket failed'
+ 
+     thread.start_new_thread(listener, ())
+     time.sleep(1)
+     connector()
+ 
+ def test_main():
+     test_rude_shutdown()
+     test_basic()
  
  if __name__ == "__main__":