Agent/examples/test_browser/test_httpx_vs_curl.py

"""
对比 curl 和 httpx 访问 CDP 端点
找出为什么 httpx 会失败
"""
import subprocess
import time
import httpx
import asyncio


def start_chrome():
    """启动 Chrome"""
    proc = subprocess.Popen([
        '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
        '--headless=new',
        '--remote-debugging-port=9222',
        '--user-data-dir=/tmp/httpx-test-cdp',
        'about:blank'
    ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    print("等待 Chrome 启动...")
    time.sleep(3)
    return proc


async def test_httpx_default():
    """测试 httpx 默认配置"""
    print("\n【测试 1】httpx 默认配置")
    print("="*60)

    try:
        async with httpx.AsyncClient() as client:
            print("发送请求...")
            resp = await client.get('http://localhost:9222/json/version')
            print(f"✅ 成功: {resp.status_code}")
            print(f"内容: {resp.text[:100]}")
            return True
    except Exception as e:
        print(f"❌ 失败: {e}")
        return False


async def test_httpx_long_timeout():
    """测试 httpx 长超时"""
    print("\n【测试 2】httpx 长超时 (30秒)")
    print("="*60)

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            print("发送请求...")
            resp = await client.get('http://localhost:9222/json/version')
            print(f"✅ 成功: {resp.status_code}")
            print(f"内容: {resp.text[:100]}")
            return True
    except Exception as e:
        print(f"❌ 失败: {e}")
        return False


async def test_httpx_no_proxy():
    """测试 httpx 禁用代理"""
    print("\n【测试 3】httpx 禁用代理")
    print("="*60)

    try:
        # 明确禁用代理
        async with httpx.AsyncClient(
            timeout=30.0,
            proxies={}  # 空字典表示不使用代理
        ) as client:
            print("发送请求...")
            resp = await client.get('http://localhost:9222/json/version')
            print(f"✅ 成功: {resp.status_code}")
            print(f"内容: {resp.text[:100]}")
            return True
    except Exception as e:
        print(f"❌ 失败: {e}")
        return False


async def test_httpx_trust_env_false():
    """测试 httpx trust_env=False"""
    print("\n【测试 4】httpx trust_env=False (忽略环境变量)")
    print("="*60)

    try:
        # trust_env=False 会忽略环境变量中的代理设置
        async with httpx.AsyncClient(
            timeout=30.0,
            trust_env=False
        ) as client:
            print("发送请求...")
            resp = await client.get('http://localhost:9222/json/version')
            print(f"✅ 成功: {resp.status_code}")
            print(f"内容: {resp.text[:100]}")
            return True
    except Exception as e:
        print(f"❌ 失败: {e}")
        return False


async def main():
    print("httpx vs curl 对比测试")
    print("="*60)

    # 启动 Chrome
    proc = start_chrome()

    # 运行测试
    results = {
        'default': await test_httpx_default(),
        'long_timeout': await test_httpx_long_timeout(),
        'no_proxy': await test_httpx_no_proxy(),
        'trust_env_false': await test_httpx_trust_env_false(),
    }

    # 清理
    print("\n清理...")
    proc.terminate()
    proc.wait(timeout=5)
    subprocess.run(['rm', '-rf', '/tmp/httpx-test-cdp'],
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # 总结
    print("\n" + "="*60)
    print("测试总结")
    print("="*60)
    for name, result in results.items():
        status = '✅ 成功' if result else '❌ 失败'
        print(f"{name:20s}: {status}")
    print("="*60)

    # 给出建议
    if results['trust_env_false'] and not results['default']:
        print("\n💡 发现问题：")
        print("httpx 默认会读取环境变量中的代理设置")
        print("即使关闭了 VPN，环境变量可能仍然存在")
        print("\n解决方案：")
        print("在 browser-use 初始化时设置 trust_env=False")


if __name__ == "__main__":
    asyncio.run(main())