<s> """<STR_LIT>""" <EOL> import base64 <EOL> import pytest <EOL> import h2 . connection <EOL> import h2 . errors <EOL> import h2 . events <EOL> import h2 . exceptions <EOL> class TestClientUpgrade ( object ) : <EOL> """<STR_LIT>""" <EOL> example_request_headers = [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:/>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:GET>' ) , <EOL> ] <EOL> example_response_headers = [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) <EOL> ] <EOL> def test_returns_http2_settings ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> conn = h2 . connection . H2Connection ( ) <EOL> data = conn . initiate_upgrade_connection ( ) <EOL> assert not data . endswith ( b'<STR_LIT:=>' ) <EOL> decoded_frame = base64 . urlsafe_b64decode ( data ) <EOL> expected_frame = frame_factory . build_settings_frame ( <EOL> settings = conn . local_settings <EOL> ) <EOL> assert decoded_frame == expected_frame . serialize ( ) <EOL> def test_emits_preamble ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> conn = h2 . connection . H2Connection ( ) <EOL> conn . initiate_upgrade_connection ( ) <EOL> data = conn . data_to_send ( ) <EOL> assert data . startswith ( frame_factory . preamble ( ) ) <EOL> data = data [ len ( frame_factory . preamble ( ) ) : ] <EOL> expected_frame = frame_factory . build_settings_frame ( <EOL> settings = conn . local_settings <EOL> ) <EOL> assert data == expected_frame . serialize ( ) <EOL> def test_can_receive_response ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> f1 = frame_factory . build_headers_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> headers = self . example_response_headers , <EOL> ) <EOL> f2 = frame_factory . build_data_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> data = b'<STR_LIT>' , <EOL> flags = [ '<STR_LIT>' ] <EOL> ) <EOL> events = c . receive_data ( f1 . serialize ( ) + f2 . serialize ( ) ) <EOL> assert len ( events ) == <NUM_LIT:3> <EOL> assert isinstance ( events [ <NUM_LIT:0> ] , h2 . events . ResponseReceived ) <EOL> assert isinstance ( events [ <NUM_LIT:1> ] , h2 . events . DataReceived ) <EOL> assert isinstance ( events [ <NUM_LIT:2> ] , h2 . events . StreamEnded ) <EOL> assert events [ <NUM_LIT:0> ] . headers == self . example_response_headers <EOL> assert events [ <NUM_LIT:1> ] . data == b'<STR_LIT>' <EOL> assert all ( e . stream_id == <NUM_LIT:1> for e in events ) <EOL> assert not c . data_to_send ( ) <EOL> def test_can_receive_pushed_stream ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> f = frame_factory . build_push_promise_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> promised_stream_id = <NUM_LIT:2> , <EOL> headers = self . example_request_headers , <EOL> ) <EOL> events = c . receive_data ( f . serialize ( ) ) <EOL> assert len ( events ) == <NUM_LIT:1> <EOL> assert isinstance ( events [ <NUM_LIT:0> ] , h2 . events . PushedStreamReceived ) <EOL> assert events [ <NUM_LIT:0> ] . headers == self . example_request_headers <EOL> assert events [ <NUM_LIT:0> ] . parent_stream_id == <NUM_LIT:1> <EOL> assert events [ <NUM_LIT:0> ] . pushed_stream_id == <NUM_LIT:2> <EOL> def test_cannot_send_headers_stream_1 ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> with pytest . raises ( h2 . exceptions . ProtocolError ) : <EOL> c . send_headers ( stream_id = <NUM_LIT:1> , headers = self . example_request_headers ) <EOL> def test_cannot_send_data_stream_1 ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> with pytest . raises ( h2 . exceptions . ProtocolError ) : <EOL> c . send_data ( stream_id = <NUM_LIT:1> , data = b'<STR_LIT>' ) <EOL> class TestServerUpgrade ( object ) : <EOL> """<STR_LIT>""" <EOL> example_request_headers = [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:/>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:GET>' ) , <EOL> ] <EOL> example_response_headers = [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) <EOL> ] <EOL> def test_returns_nothing ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> conn = h2 . connection . H2Connection ( client_side = False ) <EOL> curl_header = "<STR_LIT>" <EOL> data = conn . initiate_upgrade_connection ( curl_header ) <EOL> assert data is None <EOL> def test_emits_preamble ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> conn = h2 . connection . H2Connection ( client_side = False ) <EOL> conn . initiate_upgrade_connection ( ) <EOL> data = conn . data_to_send ( ) <EOL> expected_frame = frame_factory . build_settings_frame ( <EOL> settings = conn . local_settings <EOL> ) <EOL> assert data == expected_frame . serialize ( ) <EOL> def test_can_send_response ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( client_side = False ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> c . send_headers ( stream_id = <NUM_LIT:1> , headers = self . example_response_headers ) <EOL> c . send_data ( stream_id = <NUM_LIT:1> , data = b'<STR_LIT>' , end_stream = True ) <EOL> f1 = frame_factory . build_headers_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> headers = self . example_response_headers , <EOL> ) <EOL> f2 = frame_factory . build_data_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> data = b'<STR_LIT>' , <EOL> flags = [ '<STR_LIT>' ] <EOL> ) <EOL> expected_data = f1 . serialize ( ) + f2 . serialize ( ) <EOL> assert c . data_to_send ( ) == expected_data <EOL> def test_can_push_stream ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( client_side = False ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> c . push_stream ( <EOL> stream_id = <NUM_LIT:1> , <EOL> promised_stream_id = <NUM_LIT:2> , <EOL> request_headers = self . example_request_headers <EOL> ) <EOL> f = frame_factory . build_push_promise_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> promised_stream_id = <NUM_LIT:2> , <EOL> headers = self . example_request_headers , <EOL> ) <EOL> assert c . data_to_send ( ) == f . serialize ( ) <EOL> def test_cannot_receive_headers_stream_1 ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( client_side = False ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . receive_data ( frame_factory . preamble ( ) ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> f = frame_factory . build_headers_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> headers = self . example_request_headers , <EOL> ) <EOL> c . receive_data ( f . serialize ( ) ) <EOL> expected_frame = frame_factory . build_rst_stream_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> error_code = h2 . errors . STREAM_CLOSED , <EOL> ) <EOL> assert c . data_to_send ( ) == expected_frame . serialize ( ) <EOL> def test_cannot_receive_data_stream_1 ( self , frame_factory ) : <EOL> """<STR_LIT>""" <EOL> c = h2 . connection . H2Connection ( client_side = False ) <EOL> c . initiate_upgrade_connection ( ) <EOL> c . receive_data ( frame_factory . preamble ( ) ) <EOL> c . clear_outbound_data_buffer ( ) <EOL> f = frame_factory . build_data_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> data = b'<STR_LIT>' , <EOL> ) <EOL> c . receive_data ( f . serialize ( ) ) <EOL> expected_frame = frame_factory . build_rst_stream_frame ( <EOL> stream_id = <NUM_LIT:1> , <EOL> error_code = h2 . errors . STREAM_CLOSED , <EOL> ) <EOL> assert c . data_to_send ( ) == expected_frame . serialize ( ) </s>
