Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add coverage for interleave #11

Merged
merged 1 commit into from
Dec 9, 2023
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions tests/test_impl.py
Original file line number Diff line number Diff line change
@@ -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(