mirror of https://github.com/python/cpython
[3.13] gh-122133: Rework pure Python socketpair tests to avoid use of importlib.reload. (GH-122493) (#122504)
gh-122133: Rework pure Python socketpair tests to avoid use of importlib.reload. (GH-122493)
(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:
parent
4a6365c5f4
commit
c21a36112a
|
@ -592,30 +592,10 @@ if hasattr(_socket.socket, "share"):
|
||||||
return socket(0, 0, 0, info)
|
return socket(0, 0, 0, info)
|
||||||
__all__.append("fromshare")
|
__all__.append("fromshare")
|
||||||
|
|
||||||
if hasattr(_socket, "socketpair"):
|
# Origin: https://gist.github.com/4325783, by Geert Jansen. Public domain.
|
||||||
|
# This is used if _socket doesn't natively provide socketpair. It's
|
||||||
def socketpair(family=None, type=SOCK_STREAM, proto=0):
|
# always defined so that it can be patched in for testing purposes.
|
||||||
"""socketpair([family[, type[, proto]]]) -> (socket object, socket object)
|
def _fallback_socketpair(family=AF_INET, type=SOCK_STREAM, proto=0):
|
||||||
|
|
||||||
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
|
|
||||||
except NameError:
|
|
||||||
family = AF_INET
|
|
||||||
a, b = _socket.socketpair(family, type, proto)
|
|
||||||
a = socket(family, type, proto, a.detach())
|
|
||||||
b = socket(family, type, proto, b.detach())
|
|
||||||
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:
|
if family == AF_INET:
|
||||||
host = _LOCALHOST
|
host = _LOCALHOST
|
||||||
elif family == AF_INET6:
|
elif family == AF_INET6:
|
||||||
|
@ -668,6 +648,21 @@ else:
|
||||||
raise
|
raise
|
||||||
|
|
||||||
return (ssock, csock)
|
return (ssock, csock)
|
||||||
|
|
||||||
|
if hasattr(_socket, "socketpair"):
|
||||||
|
def socketpair(family=None, type=SOCK_STREAM, proto=0):
|
||||||
|
if family is None:
|
||||||
|
try:
|
||||||
|
family = AF_UNIX
|
||||||
|
except NameError:
|
||||||
|
family = AF_INET
|
||||||
|
a, b = _socket.socketpair(family, type, proto)
|
||||||
|
a = socket(family, type, proto, a.detach())
|
||||||
|
b = socket(family, type, proto, b.detach())
|
||||||
|
return a, b
|
||||||
|
|
||||||
|
else:
|
||||||
|
socketpair = _fallback_socketpair
|
||||||
__all__.append("socketpair")
|
__all__.append("socketpair")
|
||||||
|
|
||||||
socketpair.__doc__ = """socketpair([family[, type[, proto]]]) -> (socket object, socket object)
|
socketpair.__doc__ = """socketpair([family[, type[, proto]]]) -> (socket object, socket object)
|
||||||
|
|
|
@ -4861,7 +4861,6 @@ class BasicSocketPairTest(SocketPairTest):
|
||||||
|
|
||||||
|
|
||||||
class PurePythonSocketPairTest(SocketPairTest):
|
class PurePythonSocketPairTest(SocketPairTest):
|
||||||
|
|
||||||
# Explicitly use socketpair AF_INET or AF_INET6 to ensure that is the
|
# Explicitly use socketpair AF_INET or AF_INET6 to ensure that is the
|
||||||
# code path we're using regardless platform is the pure python one where
|
# code path we're using regardless platform is the pure python one where
|
||||||
# `_socket.socketpair` does not exist. (AF_INET does not work with
|
# `_socket.socketpair` does not exist. (AF_INET does not work with
|
||||||
|
@ -4876,28 +4875,21 @@ class PurePythonSocketPairTest(SocketPairTest):
|
||||||
# Local imports in this class make for easy security fix backporting.
|
# Local imports in this class make for easy security fix backporting.
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
import _socket
|
if hasattr(_socket, "socketpair"):
|
||||||
self._orig_sp = getattr(_socket, 'socketpair', None)
|
self._orig_sp = socket.socketpair
|
||||||
if self._orig_sp is not None:
|
|
||||||
# This forces the version using the non-OS provided socketpair
|
# This forces the version using the non-OS provided socketpair
|
||||||
# emulation via an AF_INET socket in Lib/socket.py.
|
# emulation via an AF_INET socket in Lib/socket.py.
|
||||||
del _socket.socketpair
|
socket.socketpair = socket._fallback_socketpair
|
||||||
import importlib
|
|
||||||
global socket
|
|
||||||
socket = importlib.reload(socket)
|
|
||||||
else:
|
else:
|
||||||
pass # This platform already uses the non-OS provided version.
|
# This platform already uses the non-OS provided version.
|
||||||
|
self._orig_sp = None
|
||||||
super().setUp()
|
super().setUp()
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
super().tearDown()
|
super().tearDown()
|
||||||
import _socket
|
|
||||||
if self._orig_sp is not None:
|
if self._orig_sp is not None:
|
||||||
# Restore the default socket.socketpair definition.
|
# Restore the default socket.socketpair definition.
|
||||||
_socket.socketpair = self._orig_sp
|
socket.socketpair = self._orig_sp
|
||||||
import importlib
|
|
||||||
global socket
|
|
||||||
socket = importlib.reload(socket)
|
|
||||||
|
|
||||||
def test_recv(self):
|
def test_recv(self):
|
||||||
msg = self.serv.recv(1024)
|
msg = self.serv.recv(1024)
|
||||||
|
|
Loading…
Reference in New Issue