<s> """<STR_LIT>""" <EOL> import threading <EOL> import socket <EOL> import sys <EOL> from hyper import HTTP20Connection <EOL> from hyper . compat import ssl <EOL> from hyper . http11 . connection import HTTP11Connection <EOL> from hpack . hpack import Encoder <EOL> from hpack . huffman import HuffmanEncoder <EOL> from hpack . huffman_constants import ( <EOL> REQUEST_CODES , REQUEST_CODES_LENGTH <EOL> ) <EOL> from hyper . tls import NPN_PROTOCOL <EOL> class SocketServerThread ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> socket_handler , <EOL> host = '<STR_LIT:localhost>' , <EOL> ready_event = None , <EOL> h2 = True , <EOL> secure = True ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . socket_handler = socket_handler <EOL> self . host = host <EOL> self . secure = secure <EOL> self . ready_event = ready_event <EOL> self . daemon = True <EOL> if self . secure : <EOL> self . cxt = ssl . SSLContext ( ssl . PROTOCOL_SSLv23 ) <EOL> if ssl . HAS_NPN and h2 : <EOL> self . cxt . set_npn_protocols ( [ NPN_PROTOCOL ] ) <EOL> self . cxt . load_cert_chain ( certfile = '<STR_LIT>' , <EOL> keyfile = '<STR_LIT>' ) <EOL> def _start_server ( self ) : <EOL> sock = socket . socket ( ) <EOL> if sys . platform != '<STR_LIT:win32>' : <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> if self . secure : <EOL> sock = self . cxt . wrap_socket ( sock , server_side = True ) <EOL> sock . bind ( ( self . host , <NUM_LIT:0> ) ) <EOL> self . port = sock . getsockname ( ) [ <NUM_LIT:1> ] <EOL> sock . listen ( <NUM_LIT:1> ) <EOL> if self . ready_event : <EOL> self . ready_event . set ( ) <EOL> self . socket_handler ( sock ) <EOL> sock . close ( ) <EOL> def _wrap_socket ( self , sock ) : <EOL> raise NotImplementedError ( ) <EOL> def run ( self ) : <EOL> self . server = self . _start_server ( ) <EOL> class SocketLevelTest ( object ) : <EOL> """<STR_LIT>""" <EOL> def set_up ( self , secure = True , proxy = False ) : <EOL> self . host = None <EOL> self . port = None <EOL> self . secure = secure if not proxy else False <EOL> self . proxy = proxy <EOL> self . server_thread = None <EOL> def _start_server ( self , socket_handler ) : <EOL> """<STR_LIT>""" <EOL> ready_event = threading . Event ( ) <EOL> self . server_thread = SocketServerThread ( <EOL> socket_handler = socket_handler , <EOL> ready_event = ready_event , <EOL> h2 = self . h2 , <EOL> secure = self . secure <EOL> ) <EOL> self . server_thread . start ( ) <EOL> ready_event . wait ( ) <EOL> self . host = self . server_thread . host <EOL> self . port = self . server_thread . port <EOL> self . secure = self . server_thread . secure <EOL> def get_connection ( self ) : <EOL> if self . h2 : <EOL> if not self . proxy : <EOL> return HTTP20Connection ( self . host , self . port , self . secure ) <EOL> else : <EOL> return HTTP20Connection ( '<STR_LIT>' , secure = self . secure , <EOL> proxy_host = self . host , <EOL> proxy_port = self . port ) <EOL> else : <EOL> if not self . proxy : <EOL> return HTTP11Connection ( self . host , self . port , self . secure ) <EOL> else : <EOL> return HTTP11Connection ( '<STR_LIT>' , secure = self . secure , <EOL> proxy_host = self . host , <EOL> proxy_port = self . port ) <EOL> def get_encoder ( self ) : <EOL> """<STR_LIT>""" <EOL> e = Encoder ( ) <EOL> e . huffman_coder = HuffmanEncoder ( REQUEST_CODES , REQUEST_CODES_LENGTH ) <EOL> return e <EOL> def tear_down ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server_thread . join ( <NUM_LIT:0.1> ) </s>
