1import errno 2import unittest 3 4from test_sslsocket import ServerAcceptor 5 6import _import_local_thrift # noqa 7 8from thrift.transport.TSocket import TServerSocket 9from thrift.transport.TSocket import TSocket 10from thrift.transport.TTransport import TTransportException 11 12 13class TSocketTest(unittest.TestCase): 14 def test_isOpen_checks_for_readability(self): 15 # https://docs.python.org/3/library/socket.html#notes-on-socket-timeouts 16 # https://docs.python.org/3/library/socket.html#socket.socket.settimeout 17 timeouts = [ 18 None, # blocking mode 19 0, # non-blocking mode 20 1.0, # timeout mode 21 ] 22 23 for timeout in timeouts: 24 acc = ServerAcceptor(TServerSocket(port=0)) 25 acc.start() 26 27 sock = TSocket(host="localhost", port=acc.port) 28 sock.open() 29 sock.setTimeout(timeout) 30 31 # the socket shows as open immediately after connecting 32 self.assertTrue(sock.isOpen()) 33 34 # and remains open during usage 35 sock.write(b"hello") 36 self.assertTrue(sock.isOpen()) 37 while True: 38 try: 39 sock.read(5) 40 except TTransportException as exc: 41 if exc.inner.errno == errno.EAGAIN: 42 # try again when we're in non-blocking mode 43 continue 44 raise 45 break 46 self.assertTrue(sock.isOpen()) 47 48 # once the server side closes, it no longer shows open 49 acc.client.close() # this also blocks until the other thread is done 50 acc.close() 51 self.assertFalse(sock.isOpen()) 52 53 sock.close() 54 55 56if __name__ == "__main__": 57 unittest.main() 58