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.

Sign up for GitHub

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 bind failure with local addresses #15

Merged
merged 1 commit into from
Dec 9, 2023
Merged

feat: add coverage for bind failure with local addresses #15

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
200 changes: 187 additions & 13 deletions tests/test_impl.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ def patch_socket(f):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_single_addr_info_errors(m_socket: ModuleType) -> None:
async def test__single_addr_info_errors(m_socket: ModuleType) -> None:
idx = -1
errors = ["err1", "err2"]

Expand All @@ -68,7 +68,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_single_addr_success(m_socket: ModuleType) -> None:
async def test__single_addr_success(m_socket: ModuleType) -> None:
mock_socket = mock.MagicMock(
family=socket.AF_INET,
type=socket.SOCK_STREAM,
Expand Down Expand Up @@ -96,7 +96,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_multiple_addr_success_second_one(
async def test__multiple_addr_success_second_one(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -139,7 +139,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_multiple_addr_success_second_one_happy_eyeballs(
async def test__multiple_addr_success_second_one_happy_eyeballs(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -184,7 +184,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_multiple_addr_all_fail_happy_eyeballs(
async def test__multiple_addr_all_fail_happy_eyeballs(
m_socket: ModuleType,
) -> None:
mock.MagicMock(
Expand Down Expand Up @@ -225,7 +225,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv6_and_ipv4_happy_eyeballs_ipv6_fails(
async def test__ipv6_and_ipv4_happy_eyeballs_ipv6_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -268,7 +268,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv6_and_ipv4_happy_eyeballs_ipv4_fails(
async def test__ipv6_and_ipv4_happy_eyeballs_ipv4_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -313,7 +313,7 @@ def _socket(*args, **kw):

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv6_and_ipv4_happy_eyeballs_first_ipv6_fails(
async def test__ipv6_and_ipv4_happy_eyeballs_first_ipv6_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -374,7 +374,7 @@ async def _sock_connect(

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv64_happy_eyeballs_interleave_2_first_ipv6_fails(
async def test__ipv64_happy_eyeballs_interleave_2_first_ipv6_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -437,7 +437,7 @@ async def _sock_connect(

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv6_only_happy_eyeballs_first_ipv6_fails(
async def test__ipv6_only_happy_eyeballs_first_ipv6_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -491,7 +491,7 @@ async def _sock_connect(

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv64_laddr_eyeballs_interleave_2_first_ipv6_fails(
async def test__ipv64_laddr_eyeballs_interleave_2_first_ipv6_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -568,7 +568,7 @@ async def _sock_connect(

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv64_laddr_both__eyeballs_first_ipv6_fails(
async def test__ipv64_laddr_both__eyeballs_first_ipv6_fails(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down Expand Up @@ -650,7 +650,181 @@ async def _sock_connect(

@pytest.mark.asyncio
@patch_socket
async def test_start_connection_ipv64_laddr_eyeballs_ipv4_only_tried(
async def test__ipv64_laddr_bind_fails_eyeballs_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])
if kw["family"] == socket.AF_INET:
mock_socket.bind.side_effect = OSError("bind fail")

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]
local_addr_infos = [
(
socket.AF_INET6,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"",
("::1", 0, 0, 0),
),
(
socket.AF_INET,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"",
("127.0.0.1", 0),
),
]
loop = asyncio.get_running_loop()
with mock.patch.object(loop, "sock_connect", _sock_connect), pytest.raises(
OSError, match="ipv6 fail"
):
assert (
await start_connection(
addr_info,
happy_eyeballs_delay=0.3,
interleave=1,
local_addr_infos=local_addr_infos,
)
== mock_socket
)

# We only tried IPv6 since bind to IPv4 failed
assert create_calls == [("dead:beef::", 80, 0, 0)]


@pytest.mark.asyncio
@patch_socket
async def test_ipv64_laddr_bind_fails_eyeballs_interleave_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])
if kw["family"] == socket.AF_INET:
mock_socket.bind.side_effect = OSError("bind fail")

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]
local_addr_infos = [
(
socket.AF_INET6,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"",
("::1", 0, 0, 0),
),
(
socket.AF_INET,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"",
("127.0.0.1", 0),
),
]
loop = asyncio.get_running_loop()
with mock.patch.object(loop, "sock_connect", _sock_connect):
assert (
await start_connection(
addr_info,
happy_eyeballs_delay=0.3,
interleave=2,
local_addr_infos=local_addr_infos,
)
== mock_socket
)

# IPv6 is tried first and fails, which means IPv4 is tried next but the laddr
# build fails so we move on to the next IPv6 and it succeeds
assert create_calls == [("dead:beef::", 80, 0, 0), ("dead:aaaa::", 80, 0, 0)]
assert mock_socket.family == socket.AF_INET6


@pytest.mark.asyncio
@patch_socket
async def test__ipv64_laddr_eyeballs_ipv4_only_tried(
m_socket: ModuleType,
) -> None:
mock_socket = mock.MagicMock(
Expand Down
Loading