Method test_correct_header_line

test/contrib/test_socks.py:314–347

(self)

            assert response.headers["Server"] == "SocksTestServer"

    def test_correct_header_line(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 == b"example.com"
            assert port == 80
            with pytest.raises(StopIteration):
                handler.send(True)

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

            assert buf.startswith(b"GET / HTTP/1.1")
            assert b"Host: example.com" in buf

            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"socks5h://{self.host}:{self.port}"
        with socks.SOCKSProxyManager(proxy_url) as pm:
            response = pm.request("GET", "http://example.com")
            assert response.status == 200

    def test_connection_timeouts(self) -> None:
        event = threading.Event()