Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-30 21:21:22 +00:00
Code Issues Projects Releases Packages Wiki Activity

test_rude_shutdown(): Rewrote to use proper thread

Browse source 
synchronization and termination.
This commit is contained in:
Tim Peters 2006-01-24 22:44:54 +00:00
parent 92037a15a9
commit 2a4712dc80
1 changed files with 19 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

30
Lib/test/test_socket_ssl.py
View file

@ -2,7 +2,6 @@
from test import test_support from test import test_support
import socket import socket
import time
# Optionally test SSL support. This requires the 'network' resource as given # Optionally test SSL support. This requires the 'network' resource as given
# on the regrtest command line. # on the regrtest command line.
@ -29,34 +28,43 @@ def test_basic():
def test_rude_shutdown(): def test_rude_shutdown():
try: try:
import thread import threading
except ImportError: except ImportError:
return return
# some random port to connect to # Some random port to connect to.
PORT = 9934 PORT = 9934
listener_gone = threading.Event()
# `listener` runs in a thread. It opens a socket listening on PORT, and
# sits in an accept() until the main thread connects. Then it rudely
# closes the socket, and sets Event `listener_gone` to let the main thread
# know the socket is gone.
def listener(): def listener():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s = socket.socket()
s.bind(('', PORT)) s.bind(('', PORT))
s.listen(5) s.listen(5)
s.accept() s.accept()
del s s = None # reclaim the socket object, which also closes it
thread.exit() listener_gone.set()
def connector(): def connector():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s = socket.socket()
s.connect(('localhost', PORT)) s.connect(('localhost', PORT))
listener_gone.wait()
try: try:
ssl_sock = socket.ssl(s) ssl_sock = socket.ssl(s)
except socket.sslerror: except socket.sslerror:
pass pass
else: else:
raise test_support.TestFailed, \ raise test_support.TestFailed(
'connecting to closed SSL socket failed' 'connecting to closed SSL socket should have failed')
thread.start_new_thread(listener, ()) t = threading.Thread(target=listener)
time.sleep(1) t.start()
connector() connector()
t.join()
def test_main(): def test_main():
if not hasattr(socket, "ssl"): if not hasattr(socket, "ssl"):