(self)
| 3668 | |
| 3669 | @requireAttrs(socket, "MSG_PEEK") |
| 3670 | def testRecvmsgPeek(self): |
| 3671 | # Check that MSG_PEEK in flags enables examination of pending |
| 3672 | # data without consuming it. |
| 3673 | |
| 3674 | # Receive part of data with MSG_PEEK. |
| 3675 | msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, |
| 3676 | len(MSG) - 3, 0, |
| 3677 | socket.MSG_PEEK) |
| 3678 | self.assertEqual(msg, MSG[:-3]) |
| 3679 | self.checkRecvmsgAddress(addr, self.cli_addr) |
| 3680 | self.assertEqual(ancdata, []) |
| 3681 | # Ignoring MSG_TRUNC here (so this test is the same for stream |
| 3682 | # and datagram sockets). Some wording in POSIX seems to |
| 3683 | # suggest that it needn't be set when peeking, but that may |
| 3684 | # just be a slip. |
| 3685 | self.checkFlags(flags, eor=False, |
| 3686 | ignore=getattr(socket, "MSG_TRUNC", 0)) |
| 3687 | |
| 3688 | # Receive all data with MSG_PEEK. |
| 3689 | msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, |
| 3690 | len(MSG), 0, |
| 3691 | socket.MSG_PEEK) |
| 3692 | self.assertEqual(msg, MSG) |
| 3693 | self.checkRecvmsgAddress(addr, self.cli_addr) |
| 3694 | self.assertEqual(ancdata, []) |
| 3695 | self.checkFlags(flags, eor=True) |
| 3696 | |
| 3697 | # Check that the same data can still be received normally. |
| 3698 | msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG)) |
| 3699 | self.assertEqual(msg, MSG) |
| 3700 | self.checkRecvmsgAddress(addr, self.cli_addr) |
| 3701 | self.assertEqual(ancdata, []) |
| 3702 | self.checkFlags(flags, eor=True) |
| 3703 | |
| 3704 | @testRecvmsgPeek.client_skip |
| 3705 | def _testRecvmsgPeek(self): |
nothing calls this directly
no test coverage detected