mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 08:59:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

[3.9] gh-122133: Rework pure Python socketpair tests to avoid use of importlib.reload. (GH-122493) (GH-122508)

Browse source 
(cherry picked from commit f071f01b7b)

Co-authored-by: Russell Keith-Magee <russell@keith-magee.com>
Co-authored-by: Gregory P. Smith <greg@krypto.org>
This commit is contained in:
Miss Islington (bot) 2024-08-02 15:10:52 +02:00 • committed by GitHub
parent 06fa244666
commit 3f5d9d12c7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 64 additions and 77 deletions
Download patch file Download diff file Expand all files Collapse all files

121
Lib/socket.py
View file

@ -588,16 +588,65 @@ if hasattr(_socket.socket, "share"):
return socket(0, 0, 0, info)
__all__.append("fromshare")
# Origin: https://gist.github.com/4325783, by Geert Jansen. Public domain.
# This is used if _socket doesn't natively provide socketpair. It's
# always defined so that it can be patched in for testing purposes.
def _fallback_socketpair(family=AF_INET, type=SOCK_STREAM, proto=0):
if family == AF_INET:
host = _LOCALHOST
elif family == AF_INET6:
host = _LOCALHOST_V6
else:
raise ValueError("Only AF_INET and AF_INET6 socket address families "
"are supported")
if type != SOCK_STREAM:
raise ValueError("Only SOCK_STREAM socket type is supported")
if proto != 0:
raise ValueError("Only protocol zero is supported")
# We create a connected TCP socket. Note the trick with
# setblocking(False) that prevents us from having to create a thread.
lsock = socket(family, type, proto)
try:
lsock.bind((host, 0))
lsock.listen()
# On IPv6, ignore flow_info and scope_id
addr, port = lsock.getsockname()[:2]
csock = socket(family, type, proto)
try:
csock.setblocking(False)
try:
csock.connect((addr, port))
except (BlockingIOError, InterruptedError):
pass
csock.setblocking(True)
ssock, _ = lsock.accept()
except:
csock.close()
raise
finally:
lsock.close()
# Authenticating avoids using a connection from something else
# able to connect to {host}:{port} instead of us.
# We expect only AF_INET and AF_INET6 families.
try:
if (
ssock.getsockname() != csock.getpeername()
or csock.getsockname() != ssock.getpeername()
):
raise ConnectionError("Unexpected peer connection")
except:
# getsockname() and getpeername() can fail
# if either socket isn't connected.
ssock.close()
csock.close()
raise
return (ssock, csock)
if hasattr(_socket, "socketpair"):
def socketpair(family=None, type=SOCK_STREAM, proto=0):
"""socketpair([family[, type[, proto]]]) -> (socket object, socket object)
Create a pair of socket objects from the sockets returned by the platform
socketpair() function.
The arguments are the same as for socket() except the default family is
AF_UNIX if defined on the platform; otherwise, the default is AF_INET.
"""
if family is None:
try:
family = AF_UNIX
@ -609,61 +658,7 @@ if hasattr(_socket, "socketpair"):
return a, b
else:
# Origin: https://gist.github.com/4325783, by Geert Jansen. Public domain.
def socketpair(family=AF_INET, type=SOCK_STREAM, proto=0):
if family == AF_INET:
host = _LOCALHOST
elif family == AF_INET6:
host = _LOCALHOST_V6
else:
raise ValueError("Only AF_INET and AF_INET6 socket address families "
"are supported")
if type != SOCK_STREAM:
raise ValueError("Only SOCK_STREAM socket type is supported")
if proto != 0:
raise ValueError("Only protocol zero is supported")
# We create a connected TCP socket. Note the trick with
# setblocking(False) that prevents us from having to create a thread.
lsock = socket(family, type, proto)
try:
lsock.bind((host, 0))
lsock.listen()
# On IPv6, ignore flow_info and scope_id
addr, port = lsock.getsockname()[:2]
csock = socket(family, type, proto)
try:
csock.setblocking(False)
try:
csock.connect((addr, port))
except (BlockingIOError, InterruptedError):
pass
csock.setblocking(True)
ssock, _ = lsock.accept()
except:
csock.close()
raise
finally:
lsock.close()
# Authenticating avoids using a connection from something else
# able to connect to {host}:{port} instead of us.
# We expect only AF_INET and AF_INET6 families.
try:
if (
ssock.getsockname() != csock.getpeername()
or csock.getsockname() != ssock.getpeername()
):
raise ConnectionError("Unexpected peer connection")
except:
# getsockname() and getpeername() can fail
# if either socket isn't connected.
ssock.close()
csock.close()
raise
return (ssock, csock)
socketpair = _fallback_socketpair
__all__.append("socketpair")
socketpair.__doc__ = """socketpair([family[, type[, proto]]]) -> (socket object, socket object)