(self, sock, timeout)
| 5399 | ThreadedTCPSocketTest.__init__(self, methodName=methodName) |
| 5400 | |
| 5401 | def assert_sock_timeout(self, sock, timeout): |
| 5402 | self.assertEqual(self.serv.gettimeout(), timeout) |
| 5403 | |
| 5404 | blocking = (timeout != 0.0) |
| 5405 | self.assertEqual(sock.getblocking(), blocking) |
| 5406 | |
| 5407 | if fcntl is not None: |
| 5408 | # When a Python socket has a non-zero timeout, it's switched |
| 5409 | # internally to a non-blocking mode. Later, sock.sendall(), |
| 5410 | # sock.recv(), and other socket operations use a select() call and |
| 5411 | # handle EWOULDBLOCK/EGAIN on all socket operations. That's how |
| 5412 | # timeouts are enforced. |
| 5413 | fd_blocking = (timeout is None) |
| 5414 | |
| 5415 | flag = fcntl.fcntl(sock, fcntl.F_GETFL, os.O_NONBLOCK) |
| 5416 | self.assertEqual(not bool(flag & os.O_NONBLOCK), fd_blocking) |
| 5417 | |
| 5418 | def testSetBlocking(self): |
| 5419 | # Test setblocking() and settimeout() methods |
no test coverage detected