class SingleConnectionTestServerThread(threading.Thread):
- def __init__(self, *, name, call_after_accept):
+ def __init__(self, *, name, call_after_accept, timeout=None):
self.call_after_accept = call_after_accept
self.received_data = b'' # set by .run()
self.wrap_error = None # set by .run()
self.listener = None # set by .start()
self.port = None # set by .start()
+ if timeout is None:
+ self.timeout = support.SHORT_TIMEOUT
+ else:
+ self.timeout = timeout
super().__init__(name=name)
def __enter__(self):
self.ssl_ctx.load_cert_chain(certfile=ONLYCERT, keyfile=ONLYKEY)
self.listener = socket.socket()
self.port = socket_helper.bind_port(self.listener)
- self.listener.settimeout(2.0)
+ self.listener.settimeout(self.timeout)
self.listener.listen(1)
super().start()
def run(self):
- conn, address = self.listener.accept()
- self.listener.close()
+ try:
+ conn, address = self.listener.accept()
+ except TimeoutError:
+ # on timeout, just close the listener
+ return
+ finally:
+ self.listener.close()
+
with conn:
if self.call_after_accept(conn):
return
# we're specifically trying to test. The way this test is written
# is known to work on Linux. We'll skip it anywhere else that it
# does not present as doing so.
- self.skipTest(f"Could not recreate conditions on {sys.platform}:"
- f" {err=}")
+ try:
+ self.skipTest(f"Could not recreate conditions on {sys.platform}:"
+ f" {err=}")
+ finally:
+ # gh-108342: Explicitly break the reference cycle
+ err = None
+
# If maintaining this conditional winds up being a problem.
# just turn this into an unconditional skip anything but Linux.
# The important thing is that our CI has the logic covered.
def call_after_accept(unused):
server_accept_called.set()
- if not ready_for_server_wrap_socket.wait(2.0):
+ if not ready_for_server_wrap_socket.wait(support.SHORT_TIMEOUT):
raise RuntimeError("wrap_socket event never set, test may fail.")
return False # Tell the server thread to continue.
ready_for_server_wrap_socket.set()
server.join()
+
wrap_error = server.wrap_error
- self.assertEqual(b"", server.received_data)
- self.assertIsInstance(wrap_error, OSError) # All platforms.
- self.non_linux_skip_if_other_okay_error(wrap_error)
- self.assertIsInstance(wrap_error, ssl.SSLError)
- self.assertIn("before TLS handshake with data", wrap_error.args[1])
- self.assertIn("before TLS handshake with data", wrap_error.reason)
- self.assertNotEqual(0, wrap_error.args[0])
- self.assertIsNone(wrap_error.library, msg="attr must exist")
+ server.wrap_error = None
+ try:
+ self.assertEqual(b"", server.received_data)
+ self.assertIsInstance(wrap_error, OSError) # All platforms.
+ self.non_linux_skip_if_other_okay_error(wrap_error)
+ self.assertIsInstance(wrap_error, ssl.SSLError)
+ self.assertIn("before TLS handshake with data", wrap_error.args[1])
+ self.assertIn("before TLS handshake with data", wrap_error.reason)
+ self.assertNotEqual(0, wrap_error.args[0])
+ self.assertIsNone(wrap_error.library, msg="attr must exist")
+ finally:
+ # gh-108342: Explicitly break the reference cycle
+ wrap_error = None
+ server = None
def test_preauth_data_to_tls_client(self):
+ server_can_continue_with_wrap_socket = threading.Event()
client_can_continue_with_wrap_socket = threading.Event()
def call_after_accept(conn_to_client):
+ if not server_can_continue_with_wrap_socket.wait(support.SHORT_TIMEOUT):
+ print("ERROR: test client took too long")
+
# This forces an immediate connection close via RST on .close().
set_socket_so_linger_on_with_zero_timeout(conn_to_client)
conn_to_client.send(
with socket.socket() as client:
client.connect(server.listener.getsockname())
- if not client_can_continue_with_wrap_socket.wait(2.0):
- self.fail("test server took too long.")
+ server_can_continue_with_wrap_socket.set()
+
+ if not client_can_continue_with_wrap_socket.wait(support.SHORT_TIMEOUT):
+ self.fail("test server took too long")
ssl_ctx = ssl.create_default_context()
try:
tls_client = ssl_ctx.wrap_socket(
tls_client.close()
server.join()
- self.assertEqual(b"", received_data)
- self.assertIsInstance(wrap_error, OSError) # All platforms.
- self.non_linux_skip_if_other_okay_error(wrap_error)
- self.assertIsInstance(wrap_error, ssl.SSLError)
- self.assertIn("before TLS handshake with data", wrap_error.args[1])
- self.assertIn("before TLS handshake with data", wrap_error.reason)
- self.assertNotEqual(0, wrap_error.args[0])
- self.assertIsNone(wrap_error.library, msg="attr must exist")
+ try:
+ self.assertEqual(b"", received_data)
+ self.assertIsInstance(wrap_error, OSError) # All platforms.
+ self.non_linux_skip_if_other_okay_error(wrap_error)
+ self.assertIsInstance(wrap_error, ssl.SSLError)
+ self.assertIn("before TLS handshake with data", wrap_error.args[1])
+ self.assertIn("before TLS handshake with data", wrap_error.reason)
+ self.assertNotEqual(0, wrap_error.args[0])
+ self.assertIsNone(wrap_error.library, msg="attr must exist")
+ finally:
+ # gh-108342: Explicitly break the reference cycle
+ wrap_error = None
+ server = None
def test_https_client_non_tls_response_ignored(self):
-
server_responding = threading.Event()
class SynchronizedHTTPSConnection(http.client.HTTPSConnection):
def connect(self):
+ # Call clear text HTTP connect(), not the encrypted HTTPS (TLS)
+ # connect(): wrap_socket() is called manually below.
http.client.HTTPConnection.connect(self)
+
# Wait for our fault injection server to have done its thing.
- if not server_responding.wait(1.0) and support.verbose:
+ if not server_responding.wait(support.SHORT_TIMEOUT) and support.verbose:
sys.stdout.write("server_responding event never set.")
self.sock = self._context.wrap_socket(
self.sock, server_hostname=self.host)
server_responding.set()
return True # Tell the server to stop.
+ timeout = 2.0
server = self.SingleConnectionTestServerThread(
call_after_accept=call_after_accept,
- name="non_tls_http_RST_responder")
+ name="non_tls_http_RST_responder",
+ timeout=timeout)
server.__enter__() # starts it
self.addCleanup(server.__exit__) # ... & unittest.TestCase stops it.
+
# Redundant; call_after_accept sets SO_LINGER on the accepted conn.
set_socket_so_linger_on_with_zero_timeout(server.listener)
connection = SynchronizedHTTPSConnection(
- f"localhost",
+ server.listener.getsockname()[0],
port=server.port,
context=ssl.create_default_context(),
- timeout=2.0,
+ timeout=timeout,
)
+
# There are lots of reasons this raises as desired, long before this
# test was added. Sending the request requires a successful TLS wrapped
# socket; that fails if the connection is broken. It may seem pointless
# to test this. It serves as an illustration of something that we never
# want to happen... properly not happening.
- with self.assertRaises(OSError) as err_ctx:
+ with self.assertRaises(OSError):
connection.request("HEAD", "/test", headers={"Host": "localhost"})
response = connection.getresponse()
+ server.join()
+
def setUpModule():
if support.verbose:
projects / thirdparty / Python / cpython.git / commitdiff
