Method test_preauth_data_to_tls_server

Lib/test/test_ssl.py:5589–5631

(self)

        # The important thing is that our CI has the logic covered.

    def test_preauth_data_to_tls_server(self):
        server_accept_called = threading.Event()
        ready_for_server_wrap_socket = threading.Event()

        def call_after_accept(unused):
            server_accept_called.set()
            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.

        server = self.SingleConnectionTestServerThread(
                call_after_accept=call_after_accept,
                name="preauth_data_to_tls_server")
        self.enterContext(server)  # starts it & unittest.TestCase stops it.

        with socket.socket() as client:
            client.connect(server.listener.getsockname())
            # This forces an immediate connection close via RST on .close().
            set_socket_so_linger_on_with_zero_timeout(client)
            client.setblocking(False)

            server_accept_called.wait()
            client.send(b"DELETE /data HTTP/1.0\r\n\r\n")
            client.close()  # RST

        ready_for_server_wrap_socket.set()
        server.join()

        wrap_error = server.wrap_error
        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()