Happy Eyeballs Test - Verify httpx 0.28+ behavior with IPv6

gist_happy_eyeballs_test_actual.py

#!/usr/bin/env python3
"""
Happy Eyeballs Test - Verify if Happy Eyeballs in httpx 0.28+ resolves IPv6 issues.

This tests whether we still need the manual IPv4 forcing.
Spoiler: Happy Eyeballs exists but can't help when TCP succeeds but TLS fails.
"""

import asyncio
import httpx
import time
import socket
from typing import Dict, Any

# Test URL - replace with your problematic service URL
TEST_URL = "https://api.tikapi.io"  # Or use the actual problematic service


def log_connection_info(event_name: str, info: Dict[str, Any]):
    """Log connection events from httpx."""
    if event_name == "connection.connect_tcp.started":
        print(f"🔌 Connecting to {info['host']}:{info['port']}")
    elif event_name == "connection.connect_tcp.complete":
        print(f"✅ Connected via {info.get('stream', 'unknown stream')}")
    elif event_name == "connection.connect_tcp.failed":
        print(f"❌ Connection failed: {info.get('exception', 'unknown error')}")


async def test_without_ipv4_force():
    """Test httpx without manual IPv4 forcing - relies on Happy Eyeballs."""
    print("\n=== Test 1: Without IPv4 forcing (Happy Eyeballs) ===")
    
    # Create client without any transport configuration
    async with httpx.AsyncClient() as client:
        try:
            start = time.time()
            response = await client.get(TEST_URL, timeout=30)
            elapsed = time.time() - start
            
            print(f"✅ Success! Status: {response.status_code}")
            print(f"⏱️  Time: {elapsed:.2f}s")
            
            # Try to get connection info
            if hasattr(response, '_request') and hasattr(response._request, 'stream'):
                stream = response._request.stream
                if hasattr(stream, 'connection'):
                    conn = stream.connection
                    print(f"📡 Connected via: {conn}")
            
            return True
            
        except Exception as e:
            print(f"❌ Failed: {type(e).__name__}: {e}")
            return False


async def test_with_ipv4_force():
    """Test httpx with manual IPv4 forcing (current implementation)."""
    print("\n=== Test 2: With IPv4 forcing (current implementation) ===")
    
    # Force IPv4 connections
    transport = httpx.AsyncHTTPTransport(local_address="0.0.0.0")
    async with httpx.AsyncClient(transport=transport) as client:
        try:
            start = time.time()
            response = await client.get(TEST_URL, timeout=30)
            elapsed = time.time() - start
            
            print(f"✅ Success! Status: {response.status_code}")
            print(f"⏱️  Time: {elapsed:.2f}s")
            return True
            
        except Exception as e:
            print(f"❌ Failed: {type(e).__name__}: {e}")
            return False


async def test_multiple_requests():
    """Test multiple requests to see if Happy Eyeballs learns and optimizes."""
    print("\n=== Test 3: Multiple requests (Happy Eyeballs learning) ===")
    
    async with httpx.AsyncClient() as client:
        times = []
        
        for i in range(3):
            try:
                start = time.time()
                response = await client.get(TEST_URL, timeout=30)
                elapsed = time.time() - start
                times.append(elapsed)
                
                print(f"Request {i+1}: {response.status_code} in {elapsed:.2f}s")
                
            except Exception as e:
                print(f"Request {i+1} failed: {e}")
                times.append(None)
        
        # Check if times improve (Happy Eyeballs learning)
        valid_times = [t for t in times if t is not None]
        if len(valid_times) > 1:
            if valid_times[-1] < valid_times[0]:
                print("✅ Connection times improved - Happy Eyeballs may be learning!")
            else:
                print("⚠️  Connection times didn't improve")


async def check_dns_resolution():
    """Check what addresses the hostname resolves to."""
    print("\n=== DNS Resolution Check ===")
    
    try:
        # Extract hostname from URL
        from urllib.parse import urlparse
        hostname = urlparse(TEST_URL).hostname
        
        # Get all addresses
        addr_info = socket.getaddrinfo(hostname, 443, proto=socket.IPPROTO_TCP)
        
        ipv4_addrs = []
        ipv6_addrs = []
        
        for family, type, proto, canonname, sockaddr in addr_info:
            if family == socket.AF_INET:
                ipv4_addrs.append(sockaddr[0])
            elif family == socket.AF_INET6:
                ipv6_addrs.append(sockaddr[0])
        
        print(f"Hostname: {hostname}")
        print(f"IPv4 addresses: {list(set(ipv4_addrs))}")
        print(f"IPv6 addresses: {list(set(ipv6_addrs))}")
        
    except Exception as e:
        print(f"DNS check failed: {e}")


async def main():
    """Run all tests."""
    print("Testing Happy Eyeballs in httpx 0.28.1+")
    print(f"httpx version: {httpx.__version__}")
    
    # Check DNS first
    await check_dns_resolution()
    
    # Run tests
    test1_success = await test_without_ipv4_force()
    test2_success = await test_with_ipv4_force()
    await test_multiple_requests()
    
    # Summary
    print("\n=== Summary ===")
    if test1_success and test2_success:
        print("✅ Both tests succeeded!")
        if test1_success:
            print("🎉 Happy Eyeballs appears to be working - IPv4 forcing may not be needed!")
    elif test2_success and not test1_success:
        print("⚠️  Only IPv4-forced connection worked - Happy Eyeballs may not be solving the issue")
        print("   You should keep the IPv4 forcing in place.")
    else:
        print("❌ Both tests failed - there may be other issues")


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