(self, s, nonblock=True, timeout=0.0)
| 6646 | "SOCK_NONBLOCK not defined") |
| 6647 | class NonblockConstantTest(unittest.TestCase): |
| 6648 | def checkNonblock(self, s, nonblock=True, timeout=0.0): |
| 6649 | if nonblock: |
| 6650 | self.assertEqual(s.type, socket.SOCK_STREAM) |
| 6651 | self.assertEqual(s.gettimeout(), timeout) |
| 6652 | self.assertTrue( |
| 6653 | fcntl.fcntl(s, fcntl.F_GETFL, os.O_NONBLOCK) & os.O_NONBLOCK) |
| 6654 | if timeout == 0: |
| 6655 | # timeout == 0: means that getblocking() must be False. |
| 6656 | self.assertFalse(s.getblocking()) |
| 6657 | else: |
| 6658 | # If timeout > 0, the socket will be in a "blocking" mode |
| 6659 | # from the standpoint of the Python API. For Python socket |
| 6660 | # object, "blocking" means that operations like 'sock.recv()' |
| 6661 | # will block. Internally, file descriptors for |
| 6662 | # "blocking" Python sockets *with timeouts* are in a |
| 6663 | # *non-blocking* mode, and 'sock.recv()' uses 'select()' |
| 6664 | # and handles EWOULDBLOCK/EAGAIN to enforce the timeout. |
| 6665 | self.assertTrue(s.getblocking()) |
| 6666 | else: |
| 6667 | self.assertEqual(s.type, socket.SOCK_STREAM) |
| 6668 | self.assertEqual(s.gettimeout(), None) |
| 6669 | self.assertFalse( |
| 6670 | fcntl.fcntl(s, fcntl.F_GETFL, os.O_NONBLOCK) & os.O_NONBLOCK) |
| 6671 | self.assertTrue(s.getblocking()) |
| 6672 | |
| 6673 | @support.requires_linux_version(2, 6, 28) |
| 6674 | def test_SOCK_NONBLOCK(self): |
no test coverage detected