Skip to content

Commit

Permalink
fix: test allow_hosts for unix sockets (#93)
Browse files Browse the repository at this point in the history
  • Loading branch information
miketheman committed Dec 23, 2021
1 parent 5548f81 commit 83b596a
Show file tree
Hide file tree
Showing 2 changed files with 44 additions and 10 deletions.
24 changes: 14 additions & 10 deletions pytest_socket.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,20 +56,23 @@ def socket_enabled(pytestconfig):
yield


def _is_unix_socket(family) -> bool:
try:
is_unix_socket = family == socket.AF_UNIX
except AttributeError:
# AF_UNIX not supported on Windows https://bugs.python.org/issue33408
is_unix_socket = False
return is_unix_socket


def disable_socket(allow_unix_socket=False):
""" disable socket.socket to disable the Internet. useful in testing.
"""

class GuardedSocket(socket.socket):
""" socket guard to disable socket creation (from pytest-socket) """
def __new__(cls, family=-1, type=-1, proto=-1, fileno=None):
try:
is_unix_socket = family == socket.AF_UNIX
except AttributeError:
# AF_UNIX not supported on Windows https://bugs.python.org/issue33408
is_unix_socket = False

if is_unix_socket and allow_unix_socket:
if _is_unix_socket(family) and allow_unix_socket:
return super().__new__(cls, family, type, proto, fileno)

raise SocketBlockedError()
Expand Down Expand Up @@ -130,7 +133,7 @@ def _resolve_allow_hosts(item):
hosts = mark_restrictions.args[0]
elif cli_restrictions:
hosts = cli_restrictions
socket_allow_hosts(hosts)
socket_allow_hosts(hosts, allow_unix_socket=item.config.__socket_allow_unix_socket)
return hosts


Expand All @@ -151,7 +154,7 @@ def host_from_connect_args(args):
return host_from_address(address)


def socket_allow_hosts(allowed=None):
def socket_allow_hosts(allowed=None, allow_unix_socket=False):
""" disable socket.socket.connect() to disable the Internet. useful in testing.
"""
if isinstance(allowed, str):
Expand All @@ -161,8 +164,9 @@ def socket_allow_hosts(allowed=None):

def guarded_connect(inst, *args):
host = host_from_connect_args(args)
if host and host in allowed:
if host in allowed or (_is_unix_socket(inst.family) and allow_unix_socket):
return _true_connect(inst, *args)

raise SocketConnectBlockedError(allowed, host)

socket.socket.connect = guarded_connect
Expand Down
30 changes: 30 additions & 0 deletions tests/test_combinations.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
"""Test module to express odd combinations of configurations."""
# TODO: Move this to a more common location, like `conftest.py`.
from test_async import unix_sockets_only


def test_parametrize_with_socket_enabled_and_allow_hosts(testdir, httpbin):
Expand Down Expand Up @@ -45,3 +47,31 @@ def test_remote_not_allowed_fails():
result.stdout.fnmatch_lines(
"*SocketConnectBlockedError: A test tried to use socket.socket.connect() with host*"
)


@unix_sockets_only
def test_combine_unix_and_allow_hosts(testdir, httpbin):
"""Test combination of disable, allow-unix and allow-hosts.
From https://github.com/miketheman/pytest-socket/issues/78
"""
testdir.makepyfile(
f"""
import socket
import pytest
def test_unix_connect():
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
with pytest.raises(FileNotFoundError):
sock.connect('/tmp/socket.sock')
def test_inet_connect():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('{httpbin.host}', {httpbin.port}))
"""
)
result = testdir.runpytest("--disable-socket", "--allow-unix-socket", f"--allow-hosts={httpbin.host}")
result.assert_outcomes(passed=2)

0 comments on commit 83b596a

Please sign in to comment.