(listener: socket.socket)
| 407 | done_closing = Event() |
| 408 | |
| 409 | def socket_handler(listener: socket.socket) -> None: |
| 410 | for i in 0, 1: |
| 411 | sock = listener.accept()[0] |
| 412 | |
| 413 | buf = b"" |
| 414 | while not buf.endswith(b"\r\n\r\n"): |
| 415 | buf = sock.recv(65536) |
| 416 | |
| 417 | body = f"Response {int(i)}" |
| 418 | sock.send( |
| 419 | ( |
| 420 | "HTTP/1.1 200 OK\r\n" |
| 421 | "Content-Type: text/plain\r\n" |
| 422 | "Content-Length: %d\r\n" |
| 423 | "\r\n" |
| 424 | "%s" % (len(body), body) |
| 425 | ).encode("utf-8") |
| 426 | ) |
| 427 | |
| 428 | sock.close() # simulate a server timing out, closing socket |
| 429 | done_closing.set() # let the test know it can proceed |
| 430 | |
| 431 | self._start_server(socket_handler) |
| 432 | with HTTPConnectionPool(self.host, self.port) as pool: |
nothing calls this directly
no test coverage detected