https://github.com/python/cpython/commit/45db419c3104a14007ea9efbc4bff03aef8ed10c
commit: 45db419c3104a14007ea9efbc4bff03aef8ed10c
branch: 3.11
author: Victor Stinner <[email protected]>
committer: ambv <[email protected]>
date: 2025-01-23T22:14:08+01:00
summary:
[3.11] gh-119461: Fix ThreadedVSOCKSocketStreamTest (#129171)
Fix ThreadedVSOCKSocketStreamTest: if get_cid() returns the host
address or the "any" address, use the local communication address
(loopback): VMADDR_CID_LOCAL.
On Linux 6.9, apparently, the /dev/vsock device is now available but
get_cid() returns VMADDR_CID_ANY (-1).
(cherry picked from commit e94dbe4ed83460f18bd72563c5f09f6cdc71f604)
(cherry picked from commit c750061047ee520d8299334df4b112fd983d7e48)
(cherry picked from commit cbfe3023e46b544b80ea1a38a8c900c6fb881554)
files:
M Lib/test/test_socket.py
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index cc803d8753b513..267d81294be43c 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -43,6 +43,7 @@
# test unicode string and carriage return
MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf-8')
+VMADDR_CID_LOCAL = 1
VSOCKPORT = 1234
AIX = platform.system() == "AIX"
WSL = "microsoft-standard-WSL" in platform.release()
@@ -127,8 +128,8 @@ def _have_socket_qipcrtr():
def _have_socket_vsock():
"""Check whether AF_VSOCK sockets are supported on this host."""
- ret = get_cid() is not None
- return ret
+ cid = get_cid()
+ return (cid is not None)
def _have_socket_bluetooth():
@@ -471,8 +472,6 @@ def clientTearDown(self):
@unittest.skipIf(WSL, 'VSOCK does not work on Microsoft WSL')
@unittest.skipUnless(HAVE_SOCKET_VSOCK,
'VSOCK sockets required for this test.')
[email protected](get_cid() != 2,
- "This test can only be run on a virtual guest.")
class ThreadedVSOCKSocketStreamTest(unittest.TestCase, ThreadableTest):
def __init__(self, methodName='runTest'):
@@ -494,10 +493,16 @@ def clientSetUp(self):
self.cli = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
self.addCleanup(self.cli.close)
cid = get_cid()
+ if cid in (socket.VMADDR_CID_HOST, socket.VMADDR_CID_ANY):
+ # gh-119461: Use the local communication address (loopback)
+ cid = VMADDR_CID_LOCAL
self.cli.connect((cid, VSOCKPORT))
def testStream(self):
- msg = self.conn.recv(1024)
+ try:
+ msg = self.conn.recv(1024)
+ except PermissionError as exc:
+ self.skipTest(repr(exc))
self.assertEqual(msg, MSG)
def _testStream(self):
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3/lists/python-checkins.python.org/
Member address: [email protected]
