<s> import threading <EOL> import socket <EOL> import time <EOL> import pytest <EOL> import requests <EOL> from tests . testserver . server import Server <EOL> class TestTestServer : <EOL> def test_basic ( self ) : <EOL> """<STR_LIT>""" <EOL> question = b"<STR_LIT>" <EOL> answer = b"<STR_LIT>" <EOL> def handler ( sock ) : <EOL> text = sock . recv ( <NUM_LIT:1000> ) <EOL> assert text == question <EOL> sock . sendall ( answer ) <EOL> with Server ( handler ) as ( host , port ) : <EOL> sock = socket . socket ( ) <EOL> sock . connect ( ( host , port ) ) <EOL> sock . sendall ( question ) <EOL> text = sock . recv ( <NUM_LIT:1000> ) <EOL> assert text == answer <EOL> sock . close ( ) <EOL> def test_server_closes ( self ) : <EOL> """<STR_LIT>""" <EOL> with Server . basic_response_server ( ) as ( host , port ) : <EOL> sock = socket . socket ( ) <EOL> sock . connect ( ( host , port ) ) <EOL> sock . close ( ) <EOL> with pytest . raises ( socket . error ) : <EOL> new_sock = socket . socket ( ) <EOL> new_sock . connect ( ( host , port ) ) <EOL> def test_text_response ( self ) : <EOL> """<STR_LIT>""" <EOL> server = Server . text_response_server ( <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" <EOL> ) <EOL> with server as ( host , port ) : <EOL> r = requests . get ( '<STR_LIT>' . format ( host , port ) ) <EOL> assert r . status_code == <NUM_LIT:200> <EOL> assert r . text == u'<STR_LIT>' <EOL> assert r . headers [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> def test_basic_response ( self ) : <EOL> """<STR_LIT>""" <EOL> with Server . basic_response_server ( ) as ( host , port ) : <EOL> r = requests . get ( '<STR_LIT>' . format ( host , port ) ) <EOL> assert r . status_code == <NUM_LIT:200> <EOL> assert r . text == u'<STR_LIT>' <EOL> assert r . headers [ '<STR_LIT>' ] == '<STR_LIT:0>' <EOL> def test_basic_waiting_server ( self ) : <EOL> """<STR_LIT>""" <EOL> block_server = threading . Event ( ) <EOL> with Server . basic_response_server ( wait_to_close_event = block_server ) as ( host , port ) : <EOL> sock = socket . socket ( ) <EOL> sock . connect ( ( host , port ) ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> block_server . set ( ) <EOL> def test_multiple_requests ( self ) : <EOL> """<STR_LIT>""" <EOL> requests_to_handle = <NUM_LIT:5> <EOL> server = Server . basic_response_server ( requests_to_handle = requests_to_handle ) <EOL> with server as ( host , port ) : <EOL> server_url = '<STR_LIT>' . format ( host , port ) <EOL> for _ in range ( requests_to_handle ) : <EOL> r = requests . get ( server_url ) <EOL> assert r . status_code == <NUM_LIT:200> <EOL> with pytest . raises ( requests . exceptions . ConnectionError ) : <EOL> r = requests . get ( server_url ) <EOL> def test_request_recovery ( self ) : <EOL> """<STR_LIT>""" <EOL> server = Server . basic_response_server ( requests_to_handle = <NUM_LIT:2> ) <EOL> first_request = b'<STR_LIT>' <EOL> second_request = b'<STR_LIT>' <EOL> with server as address : <EOL> sock1 = socket . socket ( ) <EOL> sock2 = socket . socket ( ) <EOL> sock1 . connect ( address ) <EOL> sock1 . sendall ( first_request ) <EOL> sock1 . close ( ) <EOL> sock2 . connect ( address ) <EOL> sock2 . sendall ( second_request ) <EOL> sock2 . close ( ) <EOL> assert server . handler_results [ <NUM_LIT:0> ] == first_request <EOL> assert server . handler_results [ <NUM_LIT:1> ] == second_request <EOL> def test_requests_after_timeout_are_not_received ( self ) : <EOL> """<STR_LIT>""" <EOL> server = Server . basic_response_server ( request_timeout = <NUM_LIT:1> ) <EOL> with server as address : <EOL> sock = socket . socket ( ) <EOL> sock . connect ( address ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> sock . close ( ) <EOL> assert server . handler_results [ <NUM_LIT:0> ] == b'<STR_LIT>' <EOL> def test_request_recovery_with_bigger_timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> server = Server . basic_response_server ( request_timeout = <NUM_LIT:3> ) <EOL> data = b'<STR_LIT>' <EOL> with server as address : <EOL> sock = socket . socket ( ) <EOL> sock . connect ( address ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> sock . sendall ( data ) <EOL> sock . close ( ) <EOL> assert server . handler_results [ <NUM_LIT:0> ] == data </s>
