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