[3.12] [3.13] gh-119461: Fix ThreadedVSOCKSocketStreamTest (GH-119465) (GH-119479...

[3.13] gh-119461: Fix ThreadedVSOCKSocketStreamTest (GH-119465) (GH-119479)

gh-119461: Fix ThreadedVSOCKSocketStreamTest (GH-119465)

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 c750061047ee520d8299334df4b112fd983d7e48)

Co-authored-by: Victor Stinner <vstinner@python.org>
(cherry picked from commit e94dbe4ed83460f18bd72563c5f09f6cdc71f604)

Co-authored-by: Victor Stinner <vstinner@python.org>

diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index 37f7fd5a04b08fa006703e730025b40735271cf6..cda956499edce7ce160e385ca6738492d6402010 100644 (file)
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -44,6 +44,7 @@ HOST = socket_helper.HOST
 # 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()
@@ -128,8 +129,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():
@@ -485,8 +486,6 @@ class ThreadedRDSSocketTest(SocketRDSTest, ThreadableTest):
 @unittest.skipIf(WSL, 'VSOCK does not work on Microsoft WSL')
 @unittest.skipUnless(HAVE_SOCKET_VSOCK,
           'VSOCK sockets required for this test.')
-@unittest.skipUnless(get_cid() != 2,
-          "This test can only be run on a virtual guest.")
 class ThreadedVSOCKSocketStreamTest(unittest.TestCase, ThreadableTest):
 
     def __init__(self, methodName='runTest'):
@@ -508,6 +507,9 @@ class ThreadedVSOCKSocketStreamTest(unittest.TestCase, ThreadableTest):
         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):