[3.10] gh-119461: Fix ThreadedVSOCKSocketStreamTest (GH-129171) (GH-1…

…29440) * [3.11] gh-119461: Fix ThreadedVSOCKSocketStreamTest (GH-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 45db419) Co-authored-by: Victor Stinner <vstinner@python.org> (cherry picked from commit e94dbe4) (cherry picked from commit c750061) (cherry picked from commit cbfe302) * gh-119461: Restore the testSocket VSOCK skipUnless removed by PR GH-119465 (GH-129561) Restore the skipUnless removed by GH-119465. This test can only pass on virtual machines, not actual machines. actual machines see: ``` self.cli.connect((cid, VSOCKPORT)) ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^ OSError: [Errno 19] No such device ``` Reproduced on (Linux) Ubuntu 24.04.1 running 6.8.0-52-generic. --------- (cherry picked from commit 2bd9f9b) Co-authored-by: Miss Islington (bot) <31488909+miss-islington@users.noreply.github.com> Co-authored-by: Victor Stinner <vstinner@python.org> Co-authored-by: Gregory P. Smith <greg@krypto.org>
python · ambv · Feb 19, 2025 · Feb 13, 2025 · Feb 18, 2025 · Feb 13, 2025
commit 4ac4704f5d3e9df7a382b9ea5a18058390120dd9
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
@@ -39,6 +39,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"

@@ -122,8 +123,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,8 @@ def clientTearDown(self):
 @unittest.skipIf(fcntl is None, "need fcntl")
 @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.")
+@unittest.skipUnless(get_cid() != 2,  # VMADDR_CID_HOST
+                     "This test can only be run on a virtual guest.")
 class ThreadedVSOCKSocketStreamTest(unittest.TestCase, ThreadableTest):

    def __init__(self, methodName='runTest'):
@@ -507,10 +508,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):