Method request_handler

test/contrib/test_socks.py:247–269

(listener: socket.socket)

    def test_basic_request(self) -> None:
        def request_handler(listener: socket.socket) -> None:
            sock = listener.accept()[0]

            handler = handle_socks5_negotiation(sock, negotiate=False)
            addr, port = next(handler)

            assert addr == "16.17.18.19"
            assert port == 80
            with pytest.raises(StopIteration):
                handler.send(True)

            while True:
                buf = sock.recv(65535)
                if buf.endswith(b"\r\n\r\n"):
                    break

            sock.sendall(
                b"HTTP/1.1 200 OK\r\n"
                b"Server: SocksTestServer\r\n"
                b"Content-Length: 0\r\n"
                b"\r\n"
            )
            sock.close()

        self._start_server(request_handler)
        proxy_url = f"socks5://{self.host}:{self.port}"