Method request_handler

test/contrib/test_socks.py:733–764

(listener: socket.socket)

    @pytest.mark.skipif(not HAS_SSL, reason=class="st">"No TLS available")
    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 == bclass="st">"localhost"
            assert port == 443
            with pytest.raises(StopIteration):
                handler.send(True)

            class="cm"># Wrap in TLS
            context = better_ssl.SSLContext(ssl.PROTOCOL_SSLv23)  class="cm"># type: ignore[misc]
            context.load_cert_chain(DEFAULT_CERTS[class="st">"certfile"], DEFAULT_CERTS[class="st">"keyfile"])
            tls = context.wrap_socket(sock, server_side=True)
            buf = bclass="st">""

            while True:
                buf += tls.recv(65535)
                if buf.endswith(bclass="st">"\r\n\r\n"):
                    break

            assert buf.startswith(bclass="st">"GET / HTTP/1.1\r\n")

            tls.sendall(
                bclass="st">"HTTP/1.1 200 OK\r\n"
                bclass="st">"Server: SocksTestServer\r\n"
                bclass="st">"Content-Length: 0\r\n"
                bclass="st">"\r\n"
            )
            tls.close()
            sock.close()

        self._start_server(request_handler)
        proxy_url = fclass="st">"socks5h://{self.host}:{self.port}"