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():
@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'):
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 = socket.VMADDR_CID_LOCAL
self.cli.connect((cid, VSOCKPORT))
def testStream(self):
socket.SO_VM_SOCKETS_BUFFER_MAX_SIZE
socket.VMADDR_CID_ANY
socket.VMADDR_PORT_ANY
+ socket.VMADDR_CID_LOCAL
socket.VMADDR_CID_HOST
socket.VM_SOCKETS_INVALID_VERSION
socket.IOCTL_VM_SOCKETS_GET_LOCAL_CID
ADD_INT_CONST(m, "SO_VM_SOCKETS_BUFFER_MAX_SIZE", 2);
ADD_INT_CONST(m, "VMADDR_CID_ANY", 0xffffffff);
ADD_INT_CONST(m, "VMADDR_PORT_ANY", 0xffffffff);
+ ADD_INT_CONST(m, "VMADDR_CID_LOCAL", 1);
ADD_INT_CONST(m, "VMADDR_CID_HOST", 2);
ADD_INT_CONST(m, "VM_SOCKETS_INVALID_VERSION", 0xffffffff);
ADD_INT_CONST(m, "IOCTL_VM_SOCKETS_GET_LOCAL_CID", _IO(7, 0xb9));