(listener: socket.socket)
| 1508 | class TestSSL(SocketDummyServerTestCase): |
| 1509 | def test_ssl_failure_midway_through_conn(self) -> None: |
| 1510 | def socket_handler(listener: socket.socket) -> None: |
| 1511 | with listener.accept()[0] as sock, sock.dup() as sock2: |
| 1512 | ssl_sock = original_ssl_wrap_socket( |
| 1513 | sock, |
| 1514 | server_side=True, |
| 1515 | keyfile=DEFAULT_CERTS["keyfile"], |
| 1516 | certfile=DEFAULT_CERTS["certfile"], |
| 1517 | ca_certs=DEFAULT_CA, |
| 1518 | ) |
| 1519 | |
| 1520 | buf = b"" |
| 1521 | while not buf.endswith(b"\r\n\r\n"): |
| 1522 | buf += ssl_sock.recv(65536) |
| 1523 | |
| 1524 | # Deliberately send from the non-SSL socket. |
| 1525 | sock2.send( |
| 1526 | b"HTTP/1.1 200 OK\r\n" |
| 1527 | b"Content-Type: text/plain\r\n" |
| 1528 | b"Content-Length: 2\r\n" |
| 1529 | b"\r\n" |
| 1530 | b"Hi" |
| 1531 | ) |
| 1532 | ssl_sock.close() |
| 1533 | |
| 1534 | self._start_server(socket_handler) |
| 1535 | with HTTPSConnectionPool(self.host, self.port, ca_certs=DEFAULT_CA) as pool: |
nothing calls this directly
no test coverage detected