diff --git a/tests/test_impl.py b/tests/test_impl.py index db79d74..ef2da76 100644 --- a/tests/test_impl.py +++ b/tests/test_impl.py @@ -372,6 +372,69 @@ async def _sock_connect( assert create_calls == [("dead:beef::", 80, 0, 0), ("107.6.106.83", 80)] +@pytest.mark.asyncio +@patch_socket +async def test_create_connection_ipv64_happy_eyeballs_interleave_2_first_ipv6_fails( + m_socket: ModuleType, +) -> None: + mock_socket = mock.MagicMock( + family=socket.AF_INET, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + fileno=mock.MagicMock(return_value=1), + ) + create_calls = [] + + def _socket(*args, **kw): + for attr in kw: + setattr(mock_socket, attr, kw[attr]) + return mock_socket + + async def _sock_connect( + sock: socket.socket, address: Tuple[str, int, int, int] + ) -> None: + create_calls.append(address) + if address[0] == "dead:beef::": + raise OSError("ipv6 fail") + + return None + + m_socket.socket = _socket # type: ignore + ipv6_addr_info = ( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("dead:beef::", 80, 0, 0), + ) + ipv6_addr_info_2 = ( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("dead:aaaa::", 80, 0, 0), + ) + ipv4_addr_info = ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("107.6.106.83", 80), + ) + addr_info = [ipv6_addr_info, ipv6_addr_info_2, ipv4_addr_info] + loop = asyncio.get_running_loop() + with mock.patch.object(loop, "sock_connect", _sock_connect): + assert ( + await create_connection(addr_info, happy_eyeballs_delay=0.3, interleave=2) + == mock_socket + ) + + # IPv6 addresses are tried first, but the first one fails so second IPv6 wins + # because interleave is 2 + assert mock_socket.family == socket.AF_INET6 + assert create_calls == [("dead:beef::", 80, 0, 0), ("dead:aaaa::", 80, 0, 0)] + + @pytest.mark.asyncio @patch_socket async def test_create_connection_ipv6_only_happy_eyeballs_first_ipv6_fails(