(self, sock, timeout)
| 5296 | ThreadedTCPSocketTest.__init__(self, methodName=methodName) |
| 5297 | |
| 5298 | def assert_sock_timeout(self, sock, timeout): |
| 5299 | self.assertEqual(self.serv.gettimeout(), timeout) |
| 5300 | |
| 5301 | blocking = (timeout != 0.0) |
| 5302 | self.assertEqual(sock.getblocking(), blocking) |
| 5303 | |
| 5304 | if fcntl is not None: |
| 5305 | # When a Python socket has a non-zero timeout, it's switched |
| 5306 | # internally to a non-blocking mode. Later, sock.sendall(), |
| 5307 | # sock.recv(), and other socket operations use a select() call and |
| 5308 | # handle EWOULDBLOCK/EGAIN on all socket operations. That's how |
| 5309 | # timeouts are enforced. |
| 5310 | fd_blocking = (timeout is None) |
| 5311 | |
| 5312 | flag = fcntl.fcntl(sock, fcntl.F_GETFL, os.O_NONBLOCK) |
| 5313 | self.assertEqual(not bool(flag & os.O_NONBLOCK), fd_blocking) |
| 5314 | |
| 5315 | def testSetBlocking(self): |
| 5316 | # Test setblocking() and settimeout() methods |
no test coverage detected