(self, s, nonblock=True, timeout=0.0)
| 6543 | "SOCK_NONBLOCK not defined") |
| 6544 | class NonblockConstantTest(unittest.TestCase): |
| 6545 | def checkNonblock(self, s, nonblock=True, timeout=0.0): |
| 6546 | if nonblock: |
| 6547 | self.assertEqual(s.type, socket.SOCK_STREAM) |
| 6548 | self.assertEqual(s.gettimeout(), timeout) |
| 6549 | self.assertTrue( |
| 6550 | fcntl.fcntl(s, fcntl.F_GETFL, os.O_NONBLOCK) & os.O_NONBLOCK) |
| 6551 | if timeout == 0: |
| 6552 | # timeout == 0: means that getblocking() must be False. |
| 6553 | self.assertFalse(s.getblocking()) |
| 6554 | else: |
| 6555 | # If timeout > 0, the socket will be in a "blocking" mode |
| 6556 | # from the standpoint of the Python API. For Python socket |
| 6557 | # object, "blocking" means that operations like 'sock.recv()' |
| 6558 | # will block. Internally, file descriptors for |
| 6559 | # "blocking" Python sockets *with timeouts* are in a |
| 6560 | # *non-blocking* mode, and 'sock.recv()' uses 'select()' |
| 6561 | # and handles EWOULDBLOCK/EAGAIN to enforce the timeout. |
| 6562 | self.assertTrue(s.getblocking()) |
| 6563 | else: |
| 6564 | self.assertEqual(s.type, socket.SOCK_STREAM) |
| 6565 | self.assertEqual(s.gettimeout(), None) |
| 6566 | self.assertFalse( |
| 6567 | fcntl.fcntl(s, fcntl.F_GETFL, os.O_NONBLOCK) & os.O_NONBLOCK) |
| 6568 | self.assertTrue(s.getblocking()) |
| 6569 | |
| 6570 | @support.requires_linux_version(2, 6, 28) |
| 6571 | def test_SOCK_NONBLOCK(self): |
no test coverage detected