Instagram Story Box

@@ -723,6 +723,10 @@ def collect_test_socket(info_add):
  if name.startswith('HAVE_')]
  copy_attributes(info_add, test_socket, 'test_socket.%s', attributes)
 
+ # Get IOCTL_VM_SOCKETS_GET_LOCAL_CID of /dev/vsock
+ cid = test_socket.get_cid()
+ info_add('test_socket.get_cid', cid)
+
 
 def collect_support(info_add):
  try:

Snap Ideas With Ruled Book

@@ -486,8 +486,8 @@ def clientTearDown(self):
 @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, # VMADDR_CID_HOST
-  "This test can only be run on a virtual guest.")
+@unittest.skipIf(get_cid() == getattr(socket, 'VMADDR_CID_HOST', 2),
+ "This test can only be run on a virtual guest.")
 class ThreadedVSOCKSocketStreamTest(unittest.TestCase, ThreadableTest):
 
  def __init__(self, methodName='runTest'):
@@ -497,7 +497,16 @@ def __init__(self, methodName='runTest'):
  def setUp(self):
  self.serv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
  self.addCleanup(self.serv.close)
- self.serv.bind((socket.VMADDR_CID_ANY, VSOCKPORT))
+ cid = get_cid()
+ if cid in (socket.VMADDR_CID_HOST, socket.VMADDR_CID_ANY):
+ cid = VMADDR_CID_LOCAL
+ try:
+ self.serv.bind((cid, VSOCKPORT))
+ except OSError as exc:
+ if exc.errno == errno.EADDRNOTAVAIL:
+ self.skipTest(f"bind() failed with {exc!r}")
+ else:
+ raise
  self.serv.listen()
  self.serverExplicitReady()
  self.serv.settimeout(support.LOOPBACK_TIMEOUT)