Method test_keepalive_disconnect

Lib/test/test_xmlrpc.py:326–363

(self)

            self.assertTrue(has_ssl)

    def test_keepalive_disconnect(self):
        class RequestHandler(http.server.BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"
            handled = False

            def do_POST(self):
                length = int(self.headers.get("Content-Length"))
                self.rfile.read(length)
                if self.handled:
                    self.close_connection = True
                    return
                response = xmlrpclib.dumps((5,), methodresponse=True)
                response = response.encode()
                self.send_response(http.HTTPStatus.OK)
                self.send_header("Content-Length", len(response))
                self.end_headers()
                self.wfile.write(response)
                self.handled = True
                self.close_connection = False

            def log_message(self, format, *args):
                # don't clobber sys.stderr
                pass

        def run_server():
            server.socket.settimeout(float(1))  # Don't hang if client fails
            server.handle_request()  # First request and attempt at second
            server.handle_request()  # Retried second request

        server = http.server.HTTPServer((socket_helper.HOST, 0), RequestHandler)
        self.addCleanup(server.server_close)
        thread = threading.Thread(target=run_server)
        thread.start()
        self.addCleanup(thread.join)
        url = "http://{}:{}/".format(*server.server_address)
        with xmlrpclib.ServerProxy(url) as p:
            self.assertEqual(p.method(), 5)
            self.assertEqual(p.method(), 5)


class SimpleXMLRPCDispatcherTestCase(unittest.TestCase):