Function test_proxy_connectivity

api_utils/routers/proxy.py:94–163

Test proxy connectivity.

(request: ProxyTestRequest)

@router.post("/test")
async def test_proxy_connectivity(request: ProxyTestRequest) -> JSONResponse:
    """Test proxy connectivity."""
    import httpx

    proxy_addr = request.address.strip()
    test_url = request.test_url

    if not proxy_addr:
        return JSONResponse(
            content=ProxyTestResult(
                success=False, message="Proxy address cannot be empty"
            ).model_dump(),
            status_code=400,
        )

    try:
        start_time = time.monotonic()
        async with httpx.AsyncClient(
            proxy=proxy_addr,
            timeout=15.0,
            follow_redirects=True,
        ) as client:
            response = await client.get(test_url)
            latency = (time.monotonic() - start_time) * 1000

            if 200 <= response.status_code < 300:
                return JSONResponse(
                    content=ProxyTestResult(
                        success=True,
                        message=f"Connection successful (HTTP {response.status_code})",
                        latency_ms=round(latency, 2),
                    ).model_dump()
                )
            else:
                return JSONResponse(
                    content=ProxyTestResult(
                        success=False,
                        message=f"HTTP Error: {response.status_code}",
                        latency_ms=round(latency, 2),
                    ).model_dump()
                )

    except httpx.ProxyError as e:
        return JSONResponse(
            content=ProxyTestResult(
                success=False, message=f"Proxy error: {e}"
            ).model_dump()
        )
    except httpx.ConnectTimeout:
        return JSONResponse(
            content=ProxyTestResult(
                success=False, message="Connection timeout"
            ).model_dump()
        )
    except httpx.ReadTimeout:
        return JSONResponse(
            content=ProxyTestResult(success=False, message="Read timeout").model_dump()
        )