From e338923fa356efa1aa61e7862f1bef9aff5e5f87 Mon Sep 17 00:00:00 2001 From: Stefan Agner Date: Mon, 14 Apr 2025 11:38:40 +0200 Subject: [PATCH 1/2] chore(test): fix resource warnings in test_net.py Make sure to close sockets after use. This fixes ResourceWarnings about unclosed sockets. --- tests/utils/test_net.py | 158 ++++++++++++++++++++-------------------- 1 file changed, 80 insertions(+), 78 deletions(-) diff --git a/tests/utils/test_net.py b/tests/utils/test_net.py index eff2befd..1eab49c3 100644 --- a/tests/utils/test_net.py +++ b/tests/utils/test_net.py @@ -165,9 +165,8 @@ def test_disable_ipv6_only_or_raise(): def _log_error(*args): errors_logged.append(args) - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) with ( - pytest.raises(OSError), + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, pytest.raises(OSError), patch.object(netutils.log, "error", _log_error), patch("socket.socket.setsockopt", side_effect=OSError), ): @@ -179,103 +178,106 @@ def _log_error(*args): ) + @pytest.mark.skipif(not hasattr(socket, "SO_REUSEPORT"), reason="System does not have SO_REUSEPORT") def test_set_so_reuseport_if_available_is_present(): """Test that setting socket.SO_REUSEPORT only OSError errno.ENOPROTOOPT is trapped.""" - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError): - netutils.set_so_reuseport_if_available(sock) + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError): + netutils.set_so_reuseport_if_available(sock) - with patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOPROTOOPT, None)): - netutils.set_so_reuseport_if_available(sock) + with patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOPROTOOPT, None)): + netutils.set_so_reuseport_if_available(sock) @pytest.mark.skipif(hasattr(socket, "SO_REUSEPORT"), reason="System has SO_REUSEPORT") def test_set_so_reuseport_if_available_not_present(): """Test that we do not try to set SO_REUSEPORT if it is not present.""" - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - with patch("socket.socket.setsockopt", side_effect=OSError): - netutils.set_so_reuseport_if_available(sock) + with ( + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, + patch("socket.socket.setsockopt", side_effect=OSError), + ): + netutils.set_so_reuseport_if_available(sock) def test_set_mdns_port_socket_options_for_ip_version(): """Test OSError with errno with EINVAL and bind address ''. from setsockopt IP_MULTICAST_TTL does not raise.""" - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: - # Should raise on EPERM always - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)): - netutils.set_mdns_port_socket_options_for_ip_version(sock, ("",), r.IPVersion.V4Only) + # Should raise on EPERM always + with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)): + netutils.set_mdns_port_socket_options_for_ip_version(sock, ("",), r.IPVersion.V4Only) - # Should raise on EINVAL always when bind address is not '' - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): - netutils.set_mdns_port_socket_options_for_ip_version(sock, ("127.0.0.1",), r.IPVersion.V4Only) + # Should raise on EINVAL always when bind address is not '' + with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): + netutils.set_mdns_port_socket_options_for_ip_version(sock, ("127.0.0.1",), r.IPVersion.V4Only) - # Should not raise on EINVAL when bind address is '' - with patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): - netutils.set_mdns_port_socket_options_for_ip_version(sock, ("",), r.IPVersion.V4Only) + # Should not raise on EINVAL when bind address is '' + with patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): + netutils.set_mdns_port_socket_options_for_ip_version(sock, ("",), r.IPVersion.V4Only) def test_add_multicast_member(caplog: pytest.LogCaptureFixture) -> None: - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - interface = "127.0.0.1" - - # EPERM should always raise - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)): - netutils.add_multicast_member(sock, interface) - - # EADDRINUSE should return False - with patch("socket.socket.setsockopt", side_effect=OSError(errno.EADDRINUSE, None)): - assert netutils.add_multicast_member(sock, interface) is False - - # EADDRNOTAVAIL should return False - with patch("socket.socket.setsockopt", side_effect=OSError(errno.EADDRNOTAVAIL, None)): - assert netutils.add_multicast_member(sock, interface) is False - - # EINVAL should return False - with patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): - assert netutils.add_multicast_member(sock, interface) is False - - # ENOPROTOOPT should return False - with patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOPROTOOPT, None)): - assert netutils.add_multicast_member(sock, interface) is False - - # ENODEV should raise for ipv4 - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.ENODEV, None)): - assert netutils.add_multicast_member(sock, interface) is False - - # ENODEV should return False for ipv6 - with patch("socket.socket.setsockopt", side_effect=OSError(errno.ENODEV, None)): - assert netutils.add_multicast_member(sock, ("2001:db8::", 1, 1)) is False # type: ignore[arg-type] - - # No IPv6 support should return False for IPv6 - with patch("socket.inet_pton", side_effect=OSError()): - assert netutils.add_multicast_member(sock, ("2001:db8::", 1, 1)) is False # type: ignore[arg-type] - - # No error should return True - with patch("socket.socket.setsockopt"): - assert netutils.add_multicast_member(sock, interface) is True - - # Ran out of IGMP memberships is forgiving and logs about igmp_max_memberships on linux - caplog.clear() - with ( - patch.object(sys, "platform", "linux"), - patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available")), - ): - assert netutils.add_multicast_member(sock, interface) is False - assert "No buffer space available" in caplog.text - assert "net.ipv4.igmp_max_memberships" in caplog.text - - # Ran out of IGMP memberships is forgiving and logs - caplog.clear() - with ( - patch.object(sys, "platform", "darwin"), - patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available")), - ): - assert netutils.add_multicast_member(sock, interface) is False - assert "No buffer space available" in caplog.text - assert "net.ipv4.igmp_max_memberships" not in caplog.text + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + interface = "127.0.0.1" + + # EPERM should always raise + with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)): + netutils.add_multicast_member(sock, interface) + + # EADDRINUSE should return False + with patch("socket.socket.setsockopt", side_effect=OSError(errno.EADDRINUSE, None)): + assert netutils.add_multicast_member(sock, interface) is False + + # EADDRNOTAVAIL should return False + with patch("socket.socket.setsockopt", side_effect=OSError(errno.EADDRNOTAVAIL, None)): + assert netutils.add_multicast_member(sock, interface) is False + + # EINVAL should return False + with patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): + assert netutils.add_multicast_member(sock, interface) is False + + # ENOPROTOOPT should return False + with patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOPROTOOPT, None)): + assert netutils.add_multicast_member(sock, interface) is False + + # ENODEV should raise for ipv4 + with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.ENODEV, None)): + assert netutils.add_multicast_member(sock, interface) is False + + # ENODEV should return False for ipv6 + with patch("socket.socket.setsockopt", side_effect=OSError(errno.ENODEV, None)): + assert netutils.add_multicast_member(sock, ("2001:db8::", 1, 1)) is False # type: ignore[arg-type] + + # No IPv6 support should return False for IPv6 + with patch("socket.inet_pton", side_effect=OSError()): + assert netutils.add_multicast_member(sock, ("2001:db8::", 1, 1)) is False # type: ignore[arg-type] + + # No error should return True + with patch("socket.socket.setsockopt"): + assert netutils.add_multicast_member(sock, interface) is True + + # Ran out of IGMP memberships is forgiving and logs about igmp_max_memberships on linux + caplog.clear() + with ( + patch.object(sys, "platform", "linux"), + patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available")), + ): + assert netutils.add_multicast_member(sock, interface) is False + assert "No buffer space available" in caplog.text + assert "net.ipv4.igmp_max_memberships" in caplog.text + + # Ran out of IGMP memberships is forgiving and logs + caplog.clear() + with ( + patch.object(sys, "platform", "darwin"), + patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available")), + ): + assert netutils.add_multicast_member(sock, interface) is False + assert "No buffer space available" in caplog.text + assert "net.ipv4.igmp_max_memberships" not in caplog.text def test_bind_raises_skips_address(): From 1ec02dc35a85cf5d73a1890c92eb3d07be7ff684 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 14 Apr 2025 09:40:54 +0000 Subject: [PATCH 2/2] chore(pre-commit.ci): auto fixes --- tests/utils/test_net.py | 35 +++++++++++++++++++++++++---------- 1 file changed, 25 insertions(+), 10 deletions(-) diff --git a/tests/utils/test_net.py b/tests/utils/test_net.py index 1eab49c3..2ed0c6f2 100644 --- a/tests/utils/test_net.py +++ b/tests/utils/test_net.py @@ -166,7 +166,8 @@ def _log_error(*args): errors_logged.append(args) with ( - socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, pytest.raises(OSError), + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, + pytest.raises(OSError), patch.object(netutils.log, "error", _log_error), patch("socket.socket.setsockopt", side_effect=OSError), ): @@ -178,7 +179,6 @@ def _log_error(*args): ) - @pytest.mark.skipif(not hasattr(socket, "SO_REUSEPORT"), reason="System does not have SO_REUSEPORT") def test_set_so_reuseport_if_available_is_present(): """Test that setting socket.SO_REUSEPORT only OSError errno.ENOPROTOOPT is trapped.""" @@ -197,7 +197,7 @@ def test_set_so_reuseport_if_available_not_present(): socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, patch("socket.socket.setsockopt", side_effect=OSError), ): - netutils.set_so_reuseport_if_available(sock) + netutils.set_so_reuseport_if_available(sock) def test_set_mdns_port_socket_options_for_ip_version(): @@ -205,13 +205,18 @@ def test_set_mdns_port_socket_options_for_ip_version(): from setsockopt IP_MULTICAST_TTL does not raise.""" with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: - # Should raise on EPERM always - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)): + with ( + pytest.raises(OSError), + patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)), + ): netutils.set_mdns_port_socket_options_for_ip_version(sock, ("",), r.IPVersion.V4Only) # Should raise on EINVAL always when bind address is not '' - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)): + with ( + pytest.raises(OSError), + patch("socket.socket.setsockopt", side_effect=OSError(errno.EINVAL, None)), + ): netutils.set_mdns_port_socket_options_for_ip_version(sock, ("127.0.0.1",), r.IPVersion.V4Only) # Should not raise on EINVAL when bind address is '' @@ -224,7 +229,10 @@ def test_add_multicast_member(caplog: pytest.LogCaptureFixture) -> None: interface = "127.0.0.1" # EPERM should always raise - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)): + with ( + pytest.raises(OSError), + patch("socket.socket.setsockopt", side_effect=OSError(errno.EPERM, None)), + ): netutils.add_multicast_member(sock, interface) # EADDRINUSE should return False @@ -244,7 +252,10 @@ def test_add_multicast_member(caplog: pytest.LogCaptureFixture) -> None: assert netutils.add_multicast_member(sock, interface) is False # ENODEV should raise for ipv4 - with pytest.raises(OSError), patch("socket.socket.setsockopt", side_effect=OSError(errno.ENODEV, None)): + with ( + pytest.raises(OSError), + patch("socket.socket.setsockopt", side_effect=OSError(errno.ENODEV, None)), + ): assert netutils.add_multicast_member(sock, interface) is False # ENODEV should return False for ipv6 @@ -263,7 +274,9 @@ def test_add_multicast_member(caplog: pytest.LogCaptureFixture) -> None: caplog.clear() with ( patch.object(sys, "platform", "linux"), - patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available")), + patch( + "socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available") + ), ): assert netutils.add_multicast_member(sock, interface) is False assert "No buffer space available" in caplog.text @@ -273,7 +286,9 @@ def test_add_multicast_member(caplog: pytest.LogCaptureFixture) -> None: caplog.clear() with ( patch.object(sys, "platform", "darwin"), - patch("socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available")), + patch( + "socket.socket.setsockopt", side_effect=OSError(errno.ENOBUFS, "No buffer space available") + ), ): assert netutils.add_multicast_member(sock, interface) is False assert "No buffer space available" in caplog.text