Skip to content
Open
Show file tree
Hide file tree
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
9 changes: 7 additions & 2 deletions src/pytest_socket/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import ipaddress
import itertools
import socket
import warnings
from collections import defaultdict
from collections.abc import Iterator
from dataclasses import dataclass, field
Expand All @@ -16,7 +17,9 @@

class SocketBlockedError(RuntimeError):
def __init__(self, *_args: Any, **_kwargs: Any) -> None:
super().__init__("A test tried to use socket.socket.")
msg = "A test tried to use socket.socket."
warnings.warn(msg, stacklevel=2)
super().__init__(msg)


class SocketConnectBlockedError(RuntimeError):
Expand All @@ -28,10 +31,12 @@ def __init__(
**_kwargs: Any,
) -> None:
allowed_str = ",".join(allowed)
super().__init__(
msg = (
"A test tried to use socket.socket.connect() "
f'with host "{host}" (allowed: "{allowed_str}").'
)
warnings.warn(msg, stacklevel=2)
super().__init__(msg)


def pytest_addoption(parser: pytest.Parser) -> None:
Expand Down
36 changes: 36 additions & 0 deletions tests/test_socket.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,42 @@ class MySocket(socket.socket):
assert_socket_blocked(result)


def test_blocked_socket_emits_warning(pytester):
"""Ensure SocketBlockedError emits a warning before raising.

This makes blocked calls visible in test output even when the
exception is caught by a bare ``except Exception`` block.
"""
pytester.makepyfile("""
import socket

def test_swallowed_socket(socket_disabled):
try:
socket.socket(socket.AF_INET, socket.SOCK_STREAM)
except Exception:
pass # swallow the error, simulating legacy code
""")
result = pytester.runpytest("-W", "always::UserWarning", "--disable-socket")
result.assert_outcomes(passed=1)
result.stdout.fnmatch_lines("*A test tried to use socket.socket.*")


def test_blocked_connect_emits_warning(pytester, httpserver):
"""Ensure SocketConnectBlockedError emits a warning before raising."""
pytester.makepyfile(f"""
import socket

def test_swallowed_connect():
try:
socket.socket().connect(('{httpserver.host}', {httpserver.port}))
except Exception:
pass # swallow the error
""")
result = pytester.runpytest("-W", "always::UserWarning", "--allow-hosts=1.2.3.4")
result.assert_outcomes(passed=1)
result.stdout.fnmatch_lines("*A test tried to use socket.socket.connect()*")


@unix_sockets_only
def test_unix_domain_sockets_blocked_with_disable_socket(pytester):
pytester.makepyfile("""
Expand Down
Loading