diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index faf326d9164e1b..d1c1da3be5ca20 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -1,7 +1,8 @@ import unittest from test import support from test.support import ( - is_apple, os_helper, refleak_helper, socket_helper, threading_helper + is_apple, os_helper, refleak_helper, socket_helper, threading_helper, + warnings_helper, ) import _thread as thread import array @@ -198,6 +199,16 @@ def socket_setdefaulttimeout(timeout): socket.setdefaulttimeout(old_timeout) +@contextlib.contextmanager +def catch_malformed_data_warning(quiet=True): + # This warning happens on macos and win, but does not always happen on linux. + with warnings_helper.check_warnings( + ("received malformed or improperly-truncated ancillary data", RuntimeWarning), + quiet=quiet, + ): + yield + + HAVE_SOCKET_CAN = _have_socket_can() HAVE_SOCKET_CAN_ISOTP = _have_socket_can_isotp() @@ -3946,8 +3957,9 @@ def checkTruncatedArray(self, ancbuf, maxdata, mindata=0): # mindata and maxdata bytes when received with buffer size # ancbuf, and that any complete file descriptor numbers are # valid. - msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, - len(MSG), ancbuf) + with catch_malformed_data_warning(): + msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, + len(MSG), ancbuf) self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr) self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC) @@ -4298,8 +4310,9 @@ def testSingleCmsgTruncInData(self): self.serv_sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_RECVHOPLIMIT, 1) self.misc_event.set() - msg, ancdata, flags, addr = self.doRecvmsg( - self.serv_sock, len(MSG), socket.CMSG_LEN(SIZEOF_INT) - 1) + with catch_malformed_data_warning(): + msg, ancdata, flags, addr = self.doRecvmsg( + self.serv_sock, len(MSG), socket.CMSG_LEN(SIZEOF_INT) - 1) self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr) @@ -4402,9 +4415,10 @@ def testSecondCmsgTruncInData(self): self.serv_sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_RECVTCLASS, 1) self.misc_event.set() - msg, ancdata, flags, addr = self.doRecvmsg( - self.serv_sock, len(MSG), - socket.CMSG_SPACE(SIZEOF_INT) + socket.CMSG_LEN(SIZEOF_INT) - 1) + with catch_malformed_data_warning(): + msg, ancdata, flags, addr = self.doRecvmsg( + self.serv_sock, len(MSG), + socket.CMSG_SPACE(SIZEOF_INT) + socket.CMSG_LEN(SIZEOF_INT) - 1) self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr)