feat: add coverage for ipv6 failure (#9)

src/aiohappyeyeballs/impl.py

@@ -8,7 +8,11 @@
 from typing import List, Optional, Sequence, Tuple, Union
 
 AddrInfoType = Tuple[
-    int, int, int, str, Union[Tuple[str, int], Tuple[str, int, int, int]]
+    Union[int, socket.AddressFamily],
+    Union[int, socket.SocketKind],
+    int,
+    str,
+    Tuple,  # type: ignore[type-arg]
 ]
 
 
@@ -40,7 +44,7 @@ async def create_connection(
     * ``sockaddr``: the socket address
 
     This method is a coroutine which will try to establish the connection
-    in the background.  When successful, the coroutine returns a
+    in the background. When successful, the coroutine returns a
     socket.
     """
     if not (current_loop := loop):

tests/test_impl.py

@@ -2,6 +2,7 @@
 import socket
 from test.test_asyncio import utils as test_utils
 from types import ModuleType
+from typing import Tuple
 from unittest import mock
 
 import pytest
@@ -122,7 +123,14 @@ def _socket(*args, **kw):
             socket.IPPROTO_TCP,
             "",
             ("107.6.106.82", 80),
-        )
+        ),
+        (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("107.6.106.83", 80),
+        ),
     ]
     loop = asyncio.get_running_loop()
     with mock.patch.object(loop, "sock_connect", return_value=None):
@@ -158,7 +166,14 @@ def _socket(*args, **kw):
             socket.IPPROTO_TCP,
             "",
             ("107.6.106.82", 80),
-        )
+        ),
+        (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("107.6.106.83", 80),
+        ),
     ]
     loop = asyncio.get_running_loop()
     with mock.patch.object(loop, "sock_connect", return_value=None):
@@ -194,8 +209,103 @@ def _socket(*args, **kw):
             socket.IPPROTO_TCP,
             "",
             ("107.6.106.82", 80),
-        )
+        ),
+        (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("107.6.106.83", 80),
+        ),
     ]
     asyncio.get_running_loop()
     with pytest.raises(OSError, match=errors[0]):
         await create_connection(addr_info, happy_eyeballs_delay=0.3)
+
+
+@pytest.mark.asyncio
+@patch_socket
+async def test_create_connection_ipv6_and_ipv4_happy_eyeballs_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),
+    )
+
+    def _socket(*args, **kw):
+        if kw["family"] == socket.AF_INET6:
+            raise OSError("ipv6 fail")
+        for attr in kw:
+            setattr(mock_socket, attr, kw[attr])
+        return mock_socket
+
+    m_socket.socket = _socket  # type: ignore
+    ipv6_addr_info = (
+        socket.AF_INET6,
+        socket.SOCK_STREAM,
+        socket.IPPROTO_TCP,
+        "",
+        ("dead:beef::", 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, ipv4_addr_info]
+    loop = asyncio.get_running_loop()
+    with mock.patch.object(loop, "sock_connect", return_value=None):
+        assert (
+            await create_connection(addr_info, happy_eyeballs_delay=0.3) == mock_socket
+        )
+    assert mock_socket.family == socket.AF_INET
+
+
+@pytest.mark.asyncio
+@patch_socket
+async def test_create_connection_ipv6_and_ipv4_happy_eyeballs_ipv4_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),
+    )
+
+    def _socket(*args, **kw):
+        if kw["family"] == socket.AF_INET:
+            raise OSError("ipv4 fail")
+        for attr in kw:
+            setattr(mock_socket, attr, kw[attr])
+        return mock_socket
+
+    m_socket.socket = _socket  # type: ignore
+    ipv6_addr: Tuple[str, int, int, int] = ("dead:beef::", 80, 0, 0)
+    ipv6_addr_info: Tuple[int, int, int, str, Tuple[str, int, int, int]] = (
+        socket.AF_INET6,
+        socket.SOCK_STREAM,
+        socket.IPPROTO_TCP,
+        "",
+        ipv6_addr,
+    )
+    ipv4_addr: Tuple[str, int] = ("107.6.106.83", 80)
+    ipv4_addr_info: Tuple[int, int, int, str, Tuple[str, int]] = (
+        socket.AF_INET,
+        socket.SOCK_STREAM,
+        socket.IPPROTO_TCP,
+        "",
+        ipv4_addr,
+    )
+    addr_info = [ipv6_addr_info, ipv4_addr_info]
+    loop = asyncio.get_running_loop()
+    with mock.patch.object(loop, "sock_connect", return_value=None):
+        assert (
+            await create_connection(addr_info, happy_eyeballs_delay=0.3) == mock_socket
+        )
+    assert mock_socket.family == socket.AF_INET6