https://github.com/python/cpython/commit/d80cbdd7084c9509d405d9091338a8bd9db90cd0
commit: d80cbdd7084c9509d405d9091338a8bd9db90cd0
branch: 3.9
author: Miss Islington (bot) <[email protected]>
committer: ambv <[email protected]>
date: 2025-02-19T14:23:10+01:00
summary:

[3.9] gh-119461: Fix ThreadedVSOCKSocketStreamTest (GH-129171) (GH-129440) 
(#130075)

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 45db419c3104a14007ea9efbc4bff03aef8ed10c)
(cherry picked from commit e94dbe4ed83460f18bd72563c5f09f6cdc71f604)
(cherry picked from commit c750061047ee520d8299334df4b112fd983d7e48)
(cherry picked from commit cbfe3023e46b544b80ea1a38a8c900c6fb881554)

---

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 2bd9f9b0547f6ffe40cd1dd964459ce11b58144d)

Co-authored-by: Victor Stinner <[email protected]>
Co-authored-by: Gregory P. Smith <[email protected]>
Co-authored-by: Petr Viktorin <[email protected]>

files:
M Lib/test/test_socket.py

diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index 1957149cac97f6..6dd21269af1146 100755
--- 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.')
[email protected](get_cid() != 2,
-          "This test can only be run on a virtual guest.")
[email protected](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):

_______________________________________________
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]

Reply via email to