(listener: socket.socket)
| 552 | |
| 553 | def test_basic_request(self) -> None: |
| 554 | def request_handler(listener: socket.socket) -> None: |
| 555 | sock = listener.accept()[0] |
| 556 | |
| 557 | handler = handle_socks4_negotiation(sock) |
| 558 | addr, port = next(handler) |
| 559 | |
| 560 | assert addr == "16.17.18.19" |
| 561 | assert port == 80 |
| 562 | with pytest.raises(StopIteration): |
| 563 | handler.send(True) |
| 564 | |
| 565 | while True: |
| 566 | buf = sock.recv(65535) |
| 567 | if buf.endswith(b"\r\n\r\n"): |
| 568 | break |
| 569 | |
| 570 | sock.sendall( |
| 571 | b"HTTP/1.1 200 OK\r\n" |
| 572 | b"Server: SocksTestServer\r\n" |
| 573 | b"Content-Length: 0\r\n" |
| 574 | b"\r\n" |
| 575 | ) |
| 576 | sock.close() |
| 577 | |
| 578 | self._start_server(request_handler) |
| 579 | proxy_url = f"socks4://{self.host}:{self.port}" |
nothing calls this directly
no test coverage detected