Function test_streaming

tests.py:568–630

Run a streaming test with the given request data.

(test_name, request_data)

            len(proxy_stats.text_content) > 0 or proxy_stats.has_tool_use)

async def test_streaming(test_name, request_data):
    """Run a streaming test with the given request data."""
    print(f"\n{'='*20} RUNNING STREAMING TEST: {test_name} {'='*20}")
    
    # Log the request data
    print(f"\nRequest data:\n{json.dumps({k: v for k, v in request_data.items() if k != 'messages'}, indent=2)}")
    
    # Make copies of the request data to avoid modifying the original
    anthropic_data = request_data.copy()
    proxy_data = request_data.copy()
    
    if not anthropic_data.get("stream"):
        anthropic_data["stream"] = True
    if not proxy_data.get("stream"):
        proxy_data["stream"] = True
    
    check_tools = "tools" in request_data
    
    try:
        # Send streaming requests
        anthropic_stats, anthropic_error = await stream_response(
            ANTHROPIC_API_URL, anthropic_headers, anthropic_data, "Anthropic"
        )
        
        proxy_stats, proxy_error = await stream_response(
            PROXY_API_URL, proxy_headers, proxy_data, "Proxy"
        )
        
        # Print statistics
        print("\n--- Anthropic Stream Statistics ---")
        anthropic_stats.summarize()
        
        print("\n--- Proxy Stream Statistics ---")
        proxy_stats.summarize()
        
        # Compare the responses
        if anthropic_error:
            print(f"\n⚠️ Anthropic stream had an error: {anthropic_error}")
            # If Anthropic errors, the test passes if proxy does anything useful
            if not proxy_error and proxy_stats.total_chunks > 0:
                print(f"\n✅ Test {test_name} passed! (Proxy worked even though Anthropic failed)")
                return True
            else:
                print(f"\n❌ Test {test_name} failed! Both streams had errors.")
                return False
        
        if proxy_error:
            print(f"\n❌ Test {test_name} failed! Proxy had an error: {proxy_error}")
            return False
        
        result = compare_stream_stats(anthropic_stats, proxy_stats)
        if result:
            print(f"\n✅ Test {test_name} passed!")
            return True
        else:
            print(f"\n❌ Test {test_name} failed!")
            return False