<s> from __future__ import absolute_import , division <EOL> import os <EOL> import socket <EOL> import pytest <EOL> from thriftpy . transport import TTransportException <EOL> from thriftpy . transport . socket import TSocket , TServerSocket <EOL> def _test_socket ( server_socket , client_socket ) : <EOL> server_socket . listen ( ) <EOL> client_socket . open ( ) <EOL> conn = server_socket . accept ( ) <EOL> buff = b"<STR_LIT>" <EOL> client_socket . write ( buff ) <EOL> buff2 = conn . read ( <NUM_LIT> ) <EOL> conn . write ( buff2 ) <EOL> buff3 = client_socket . read ( <NUM_LIT> ) <EOL> assert buff == buff2 == buff3 <EOL> conn . close ( ) <EOL> client_socket . close ( ) <EOL> server_socket . close ( ) <EOL> def test_inet_socket ( ) : <EOL> server_socket = TServerSocket ( host = "<STR_LIT:127.0.0.1>" , port = <NUM_LIT> ) <EOL> client_socket = TSocket ( host = "<STR_LIT:127.0.0.1>" , port = <NUM_LIT> ) <EOL> _test_socket ( server_socket , client_socket ) <EOL> def test_inet6_socket ( ) : <EOL> server_socket = TServerSocket ( host = "<STR_LIT>" , port = <NUM_LIT> , <EOL> socket_family = socket . AF_INET6 ) <EOL> client_socket = TSocket ( host = "<STR_LIT>" , port = <NUM_LIT> , <EOL> socket_family = socket . AF_INET6 ) <EOL> _test_socket ( server_socket , client_socket ) <EOL> def test_unix_domain_socket ( ) : <EOL> sock_file = "<STR_LIT>" <EOL> if os . path . exists ( sock_file ) : <EOL> os . unlink ( sock_file ) <EOL> os . mknod ( sock_file ) <EOL> server_socket = TServerSocket ( unix_socket = sock_file ) <EOL> client_socket = TSocket ( unix_socket = sock_file ) <EOL> _test_socket ( server_socket , client_socket ) <EOL> def test_client_socket_timeout ( ) : <EOL> server_socket = TServerSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> client_timeout = <NUM_LIT:100> ) <EOL> server_socket . listen ( ) <EOL> client_socket = TSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> connect_timeout = <NUM_LIT:10> , socket_timeout = <NUM_LIT:100> ) <EOL> client_socket . open ( ) <EOL> conn = server_socket . accept ( ) <EOL> assert client_socket . sock . gettimeout ( ) == <NUM_LIT:100> / <NUM_LIT:1000> <EOL> assert conn . sock . gettimeout ( ) == <NUM_LIT:100> / <NUM_LIT:1000> <EOL> assert server_socket . sock . gettimeout ( ) is None <EOL> conn . close ( ) <EOL> client_socket . close ( ) <EOL> server_socket . close ( ) <EOL> def test_client_socket_close ( ) : <EOL> server_socket = TServerSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> ) <EOL> server_socket . listen ( ) <EOL> client_socket = TSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> ) <EOL> client_socket . open ( ) <EOL> conn = server_socket . accept ( ) <EOL> client_socket . close ( ) <EOL> with pytest . raises ( TTransportException ) as e : <EOL> conn . read ( <NUM_LIT> ) <EOL> assert "<STR_LIT>" in e . value . message <EOL> conn . write ( b"<STR_LIT>" ) <EOL> conn . close ( ) <EOL> server_socket . close ( ) <EOL> def test_server_socket_close ( ) : <EOL> server_socket = TServerSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> ) <EOL> server_socket . listen ( ) <EOL> client_socket = TSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , socket_timeout = <NUM_LIT:100> ) <EOL> client_socket . open ( ) <EOL> conn = server_socket . accept ( ) <EOL> server_socket . close ( ) <EOL> client_socket . write ( b"<STR_LIT>" ) <EOL> with pytest . raises ( socket . timeout ) : <EOL> client_socket . read ( <NUM_LIT> ) <EOL> conn . close ( ) <EOL> server_socket . close ( ) <EOL> def test_client_socket_set_timeout ( ) : <EOL> server_socket = TServerSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , <EOL> client_timeout = <NUM_LIT:100> ) <EOL> server_socket . listen ( ) <EOL> client_socket = TSocket ( host = "<STR_LIT:localhost>" , port = <NUM_LIT> , socket_timeout = <NUM_LIT:100> ) <EOL> client_socket . open ( ) <EOL> conn = server_socket . accept ( ) <EOL> assert client_socket . sock . gettimeout ( ) == <NUM_LIT:100> / <NUM_LIT:1000> <EOL> assert conn . sock . gettimeout ( ) == <NUM_LIT:100> / <NUM_LIT:1000> <EOL> assert conn . sock . gettimeout ( ) == <NUM_LIT:100> / <NUM_LIT:1000> <EOL> client_socket . set_timeout ( <NUM_LIT:200> ) <EOL> conn . set_timeout ( <NUM_LIT:200> ) <EOL> assert client_socket . sock . gettimeout ( ) == <NUM_LIT:200> / <NUM_LIT:1000> <EOL> assert conn . sock . gettimeout ( ) == <NUM_LIT:200> / <NUM_LIT:1000> <EOL> conn . close ( ) <EOL> client_socket . close ( ) <EOL> server_socket . close ( ) </s>
