Jan 23, 2025 · May 23, 2024 · Dec 20, 2024
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
 # 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()

 def _have_socket_vsock():
    """Check whether AF_VSOCK sockets are supported on this host."""
 ret = get_cid() is not None
    returnret
 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 = 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):
Original file line number	Diff line number	Diff line change
Expand Up		@@ -43,6 +43,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"
		WSL = "microsoft-standard-WSL" in platform.release()
Expand DownExpand Up		@@ -127,8 +128,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
		returnret
		cid = get_cid()
		return(cid is not None)


		def _have_socket_bluetooth():
Expand DownExpand Up		@@ -471,8 +472,6 @@ 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,
		"This test can only be run on a virtual guest.")
		class ThreadedVSOCKSocketStreamTest(unittest.TestCase, ThreadableTest):

		def __init__(self, methodName='runTest'):
Expand All		@@ -494,10 +493,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):
Expand Down