mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-07-30 14:44:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

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

Browse source 
gh-119461: Fix ThreadedVSOCKSocketStreamTest (#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 e94dbe4ed8)
This commit is contained in:
Victor Stinner 2024-05-24 00:27:04 +02:00 • committed by GitHub
parent d98d6b1776
commit c750061047
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 6 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

10
Lib/test/test_socket.py
View file

@ -47,6 +47,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()
@ -160,8 +161,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():
@ -520,8 +521,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'):
@ -543,6 +542,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):