<s> """<STR_LIT>""" <EOL> import os <EOL> import zlib <EOL> from collections import namedtuple <EOL> from io import BytesIO , StringIO <EOL> import mock <EOL> import pytest <EOL> import hyper <EOL> from hyper . http11 . connection import HTTP11Connection <EOL> from hyper . http11 . response import HTTP11Response <EOL> from hyper . common . headers import HTTPHeaderMap <EOL> from hyper . common . exceptions import ChunkedDecodeError , ConnectionResetError <EOL> from hyper . compat import bytes , zlib_compressobj <EOL> class TestHTTP11Connection ( object ) : <EOL> def test_pycohttpparser_installs_correctly ( self ) : <EOL> if os . environ . get ( '<STR_LIT>' ) == '<STR_LIT:true>' : <EOL> import pycohttpparser <EOL> else : <EOL> with pytest . raises ( ImportError ) : <EOL> import pycohttpparser <EOL> assert True <EOL> def test_initialization_no_port ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> assert c . host == '<STR_LIT>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert not c . proxy_host <EOL> def test_initialization_inline_port ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> assert c . host == '<STR_LIT>' <EOL> assert c . port == <NUM_LIT> <EOL> assert c . secure <EOL> assert not c . proxy_host <EOL> def test_initialization_separate_port ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT:localhost>' , <NUM_LIT> ) <EOL> assert c . host == '<STR_LIT:localhost>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert not c . proxy_host <EOL> def test_can_override_security ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT:localhost>' , <NUM_LIT> , secure = False ) <EOL> assert c . host == '<STR_LIT:localhost>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert not c . proxy_host <EOL> def test_initialization_proxy ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' , proxy_host = '<STR_LIT:localhost>' ) <EOL> assert c . host == '<STR_LIT>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert c . proxy_host == '<STR_LIT:localhost>' <EOL> assert c . proxy_port == <NUM_LIT> <EOL> def test_initialization_proxy_with_inline_port ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' , proxy_host = '<STR_LIT>' ) <EOL> assert c . host == '<STR_LIT>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert c . proxy_host == '<STR_LIT:localhost>' <EOL> assert c . proxy_port == <NUM_LIT> <EOL> def test_initialization_proxy_with_separate_port ( self ) : <EOL> c = HTTP11Connection ( <EOL> '<STR_LIT>' , proxy_host = '<STR_LIT:localhost>' , proxy_port = <NUM_LIT> <EOL> ) <EOL> assert c . host == '<STR_LIT>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert c . proxy_host == '<STR_LIT:localhost>' <EOL> assert c . proxy_port == <NUM_LIT> <EOL> def test_initialization_with_ipv6_addresses_proxy_inline_port ( self ) : <EOL> c = HTTP11Connection ( <EOL> '<STR_LIT>' , proxy_host = '<STR_LIT>' <EOL> ) <EOL> assert c . host == '<STR_LIT>' <EOL> assert c . port == <NUM_LIT> <EOL> assert not c . secure <EOL> assert c . proxy_host == '<STR_LIT>' <EOL> assert c . proxy_port == <NUM_LIT> <EOL> def test_basic_request ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> c . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_iterable_header ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> c . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:test>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:test>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ) ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_invalid_header ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = DummySocket ( ) <EOL> with pytest . raises ( ValueError ) : <EOL> c . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = <NUM_LIT> ) <EOL> def test_proxy_request ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' , proxy_host = '<STR_LIT:localhost>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> c . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_request_with_bytestring_body ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> c . request ( <EOL> '<STR_LIT:POST>' , <EOL> '<STR_LIT>' , <EOL> headers = HTTPHeaderMap ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> body = b'<STR_LIT>' <EOL> ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_request_with_file_body ( self ) : <EOL> FstatRval = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> def fake_fstat ( * args ) : <EOL> return FstatRval ( <NUM_LIT:16> ) <EOL> old_fstat = hyper . http11 . connection . os . fstat <EOL> try : <EOL> hyper . http11 . connection . os . fstat = fake_fstat <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> f = DummyFile ( b'<STR_LIT>' ) <EOL> c . request ( '<STR_LIT:POST>' , '<STR_LIT>' , body = f ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> finally : <EOL> hyper . http11 . connection . os . fstat = old_fstat <EOL> def test_request_with_generator_body ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> def body ( ) : <EOL> yield b'<STR_LIT>' <EOL> yield b'<STR_LIT>' <EOL> yield b'<STR_LIT>' <EOL> c . request ( '<STR_LIT:POST>' , '<STR_LIT>' , body = body ( ) ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_content_length_overrides_generator ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> def body ( ) : <EOL> yield b'<STR_LIT>' <EOL> yield b'<STR_LIT>' <EOL> yield b'<STR_LIT>' <EOL> c . request ( <EOL> '<STR_LIT:POST>' , '<STR_LIT>' , body = body ( ) , headers = { b'<STR_LIT>' : b'<STR_LIT>' } <EOL> ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_chunked_overrides_body ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> f = DummyFile ( b'<STR_LIT>' ) <EOL> c . request ( <EOL> '<STR_LIT:POST>' , <EOL> '<STR_LIT>' , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> body = f <EOL> ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> assert received == expected <EOL> def test_get_response ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> sock . _buffer = BytesIO ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> ) <EOL> r = c . get_response ( ) <EOL> assert r . status == <NUM_LIT> <EOL> assert r . reason == b'<STR_LIT>' <EOL> assert list ( r . headers . iter_raw ( ) ) == [ <EOL> ( b'<STR_LIT>' , b'<STR_LIT>' ) , <EOL> ( b'<STR_LIT>' , b'<STR_LIT>' ) , <EOL> ( b'<STR_LIT>' , b'<STR_LIT:0>' ) <EOL> ] <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_response_short_reads ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> sock . _buffer = BytesIO ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> r = c . get_response ( ) <EOL> assert r . status == <NUM_LIT:200> <EOL> assert r . reason == b'<STR_LIT:OK>' <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT:hello>' <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT>' <EOL> def test_request_with_unicodestring_body ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = DummySocket ( ) <EOL> with pytest . raises ( ValueError ) : <EOL> c . request ( <EOL> '<STR_LIT:POST>' , <EOL> '<STR_LIT>' , <EOL> headers = HTTPHeaderMap ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> body = u'<STR_LIT>' <EOL> ) <EOL> def test_request_with_file_body_in_text_mode ( self ) : <EOL> FstatRval = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> def fake_fstat ( * args ) : <EOL> return FstatRval ( <NUM_LIT:16> ) <EOL> old_fstat = hyper . http11 . connection . os . fstat <EOL> try : <EOL> hyper . http11 . connection . os . fstat = fake_fstat <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = DummySocket ( ) <EOL> f = DummyFile ( b'<STR_LIT>' ) <EOL> f . buffer = StringIO ( u'<STR_LIT>' ) <EOL> with pytest . raises ( ValueError ) : <EOL> c . request ( '<STR_LIT:POST>' , '<STR_LIT>' , body = f ) <EOL> finally : <EOL> hyper . http11 . connection . os . fstat = old_fstat <EOL> def test_request_with_unicode_generator_body ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = DummySocket ( ) <EOL> def body ( ) : <EOL> yield u'<STR_LIT>' <EOL> yield u'<STR_LIT>' <EOL> yield u'<STR_LIT>' <EOL> with pytest . raises ( ValueError ) : <EOL> c . request ( '<STR_LIT:POST>' , '<STR_LIT>' , body = body ( ) ) <EOL> def test_content_length_overrides_generator_unicode ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = DummySocket ( ) <EOL> def body ( ) : <EOL> yield u'<STR_LIT>' <EOL> yield u'<STR_LIT>' <EOL> yield u'<STR_LIT>' <EOL> with pytest . raises ( ValueError ) : <EOL> c . request ( <EOL> '<STR_LIT:POST>' , <EOL> '<STR_LIT>' , <EOL> headers = { b'<STR_LIT>' : b'<STR_LIT>' } , <EOL> body = body ( ) <EOL> ) <EOL> def test_http_upgrade_headers_only_sent_once ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> c . _sock = sock = DummySocket ( ) <EOL> c . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> sock . queue = [ ] <EOL> c . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> received = b'<STR_LIT>' . join ( sock . queue ) <EOL> expected = ( <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> b"<STR_LIT:\r\n>" <EOL> ) <EOL> assert received == expected <EOL> def test_exception_raised_for_illegal_body_type ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> with pytest . raises ( ValueError ) as exc_info : <EOL> body = <NUM_LIT> <EOL> c . request ( <EOL> '<STR_LIT:GET>' , <EOL> '<STR_LIT>' , <EOL> body = body , <EOL> headers = { '<STR_LIT>' : str ( len ( str ( body ) ) ) } <EOL> ) <EOL> assert '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' . format ( type ( body ) ) in str ( exc_info ) <EOL> def test_exception_raised_for_illegal_elements_in_iterable_body ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> rogue_element = <NUM_LIT> <EOL> body = [ b'<STR_LIT>' , b'<STR_LIT>' , rogue_element ] <EOL> body_size = sum ( len ( bytes ( x ) ) for x in body ) <EOL> with pytest . raises ( ValueError ) as exc_info : <EOL> c . request ( <EOL> '<STR_LIT:GET>' , <EOL> '<STR_LIT>' , <EOL> body = body , <EOL> headers = { '<STR_LIT>' : str ( body_size ) } <EOL> ) <EOL> assert '<STR_LIT>' '<STR_LIT>' . format ( rogue_element ) in str ( exc_info ) <EOL> def test_exception_raised_for_filelike_body_not_returning_bytes ( self ) : <EOL> c = HTTP11Connection ( '<STR_LIT>' ) <EOL> class RogueFile ( object ) : <EOL> def read ( self , size ) : <EOL> return <NUM_LIT> <EOL> with pytest . raises ( ValueError ) as exc_info : <EOL> c . request ( <EOL> '<STR_LIT:GET>' , <EOL> '<STR_LIT>' , <EOL> body = RogueFile ( ) , <EOL> headers = { '<STR_LIT>' : str ( <NUM_LIT:10> ) } <EOL> ) <EOL> assert '<STR_LIT>' '<STR_LIT>' . format ( int ) in str ( exc_info ) <EOL> class TestHTTP11Response ( object ) : <EOL> def test_short_circuit_read ( self ) : <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT:0>' ] } , None , None ) <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_aborted_reads ( self ) : <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , None ) <EOL> with pytest . raises ( ConnectionResetError ) : <EOL> r . read ( ) <EOL> def test_read_expect_close ( self ) : <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , None ) <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_response_as_context_manager ( self ) : <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT:0>' ] } , DummySocket ( ) , None <EOL> ) <EOL> with r : <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> assert r . _sock is None <EOL> def test_response_transparently_decrypts_gzip ( self ) : <EOL> d = DummySocket ( ) <EOL> headers = { b'<STR_LIT>' : [ b'<STR_LIT>' ] , b'<STR_LIT>' : [ b'<STR_LIT>' ] } <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , None ) <EOL> c = zlib_compressobj ( wbits = <NUM_LIT> ) <EOL> body = c . compress ( b'<STR_LIT>' ) <EOL> body += c . flush ( ) <EOL> d . _buffer = BytesIO ( body ) <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_response_transparently_decrypts_real_deflate ( self ) : <EOL> d = DummySocket ( ) <EOL> headers = { <EOL> b'<STR_LIT>' : [ b'<STR_LIT>' ] , <EOL> b'<STR_LIT>' : [ b'<STR_LIT>' ] , <EOL> } <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , None ) <EOL> c = zlib_compressobj ( wbits = zlib . MAX_WBITS ) <EOL> body = c . compress ( b'<STR_LIT>' ) <EOL> body += c . flush ( ) <EOL> d . _buffer = BytesIO ( body ) <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_response_transparently_decrypts_wrong_deflate ( self ) : <EOL> c = zlib_compressobj ( wbits = - zlib . MAX_WBITS ) <EOL> body = c . compress ( b'<STR_LIT>' ) <EOL> body += c . flush ( ) <EOL> body_len = ( '<STR_LIT:%s>' % len ( body ) ) . encode ( '<STR_LIT:ascii>' ) <EOL> headers = { <EOL> b'<STR_LIT>' : [ b'<STR_LIT>' ] , b'<STR_LIT>' : [ body_len ] <EOL> } <EOL> d = DummySocket ( ) <EOL> d . _buffer = BytesIO ( body ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , None ) <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_basic_chunked_read ( self ) : <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , None <EOL> ) <EOL> data = ( <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> ) <EOL> d . _buffer = BytesIO ( data ) <EOL> results = [ <EOL> b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' <EOL> ] <EOL> for c1 , c2 in zip ( results , r . read_chunked ( ) ) : <EOL> assert c1 == c2 <EOL> assert not list ( r . read_chunked ( ) ) <EOL> def test_chunked_read_of_non_chunked ( self ) : <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT:0>' ] } , None , None ) <EOL> with pytest . raises ( ChunkedDecodeError ) : <EOL> list ( r . read_chunked ( ) ) <EOL> def test_chunked_read_aborts_early ( self ) : <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , None , None <EOL> ) <EOL> assert not list ( r . read_chunked ( ) ) <EOL> def test_response_transparently_decrypts_chunked_gzip ( self ) : <EOL> d = DummySocket ( ) <EOL> headers = { <EOL> b'<STR_LIT>' : [ b'<STR_LIT>' ] , <EOL> b'<STR_LIT>' : [ b'<STR_LIT>' ] , <EOL> } <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , None ) <EOL> c = zlib_compressobj ( wbits = <NUM_LIT> ) <EOL> body = c . compress ( b'<STR_LIT>' ) <EOL> body += c . flush ( ) <EOL> data = b'<STR_LIT>' <EOL> for index in range ( <NUM_LIT:0> , len ( body ) , <NUM_LIT:2> ) : <EOL> data += b'<STR_LIT>' + body [ index : index + <NUM_LIT:2> ] + b'<STR_LIT:\r\n>' <EOL> data += b'<STR_LIT>' <EOL> d . _buffer = BytesIO ( data ) <EOL> received_body = b'<STR_LIT>' <EOL> for chunk in r . read_chunked ( ) : <EOL> received_body += chunk <EOL> assert received_body == b'<STR_LIT>' <EOL> def test_chunked_normal_read ( self ) : <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , None ) <EOL> data = ( <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> ) <EOL> d . _buffer = BytesIO ( data ) <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_chunk_length_read ( self ) : <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , None <EOL> ) <EOL> data = ( <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> ) <EOL> d . _buffer = BytesIO ( data ) <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:15> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:20> ) == b'<STR_LIT>' <EOL> assert r . read ( ) == b'<STR_LIT>' <EOL> def test_bounded_read_expect_close_no_content_length ( self ) : <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , None ) <EOL> d . _buffer = BytesIO ( b'<STR_LIT>' ) <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT:hello>' <EOL> assert r . read ( <NUM_LIT:6> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:8> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:9> ) == b'<STR_LIT>' <EOL> assert r . _sock is None <EOL> def test_bounded_read_expect_close_with_content_length ( self ) : <EOL> headers = { b'<STR_LIT>' : [ b'<STR_LIT>' ] , b'<STR_LIT>' : [ b'<STR_LIT>' ] } <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , None ) <EOL> d . _buffer = BytesIO ( b'<STR_LIT>' ) <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT:hello>' <EOL> assert r . read ( <NUM_LIT:6> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:8> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:9> ) == b'<STR_LIT>' <EOL> assert r . _sock is None <EOL> def test_compressed_bounded_read_expect_close ( self ) : <EOL> headers = { b'<STR_LIT>' : [ b'<STR_LIT>' ] , b'<STR_LIT>' : [ b'<STR_LIT>' ] } <EOL> c = zlib_compressobj ( wbits = <NUM_LIT> ) <EOL> body = c . compress ( b'<STR_LIT>' ) <EOL> body += c . flush ( ) <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , None ) <EOL> d . _buffer = BytesIO ( body ) <EOL> response = b'<STR_LIT>' <EOL> while True : <EOL> chunk = r . read ( <NUM_LIT:15> ) <EOL> if not chunk : <EOL> break <EOL> response += chunk <EOL> assert response == b'<STR_LIT>' <EOL> assert r . _sock is None <EOL> def test_expect_close_reads_call_close_callback ( self ) : <EOL> connection = mock . MagicMock ( ) <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , connection <EOL> ) <EOL> d . _buffer = BytesIO ( b'<STR_LIT>' ) <EOL> assert r . read ( <NUM_LIT:5> ) == b'<STR_LIT:hello>' <EOL> assert r . read ( <NUM_LIT:6> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:8> ) == b'<STR_LIT>' <EOL> assert r . read ( <NUM_LIT:9> ) == b'<STR_LIT>' <EOL> assert r . _sock is None <EOL> assert connection . close . call_count == <NUM_LIT:1> <EOL> def test_expect_close_unbounded_reads_call_close_callback ( self ) : <EOL> connection = mock . MagicMock ( ) <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <EOL> <NUM_LIT:200> , '<STR_LIT:OK>' , { b'<STR_LIT>' : [ b'<STR_LIT>' ] } , d , connection <EOL> ) <EOL> d . _buffer = BytesIO ( b'<STR_LIT>' ) <EOL> r . read ( ) <EOL> assert r . _sock is None <EOL> assert connection . close . call_count == <NUM_LIT:1> <EOL> def test_content_length_expect_close_reads_call_close_callback ( self ) : <EOL> connection = mock . MagicMock ( ) <EOL> headers = { b'<STR_LIT>' : [ b'<STR_LIT>' ] , b'<STR_LIT>' : [ b'<STR_LIT>' ] } <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , connection ) <EOL> d . _buffer = BytesIO ( b'<STR_LIT>' ) <EOL> r . read ( ) <EOL> assert r . _sock is None <EOL> assert connection . close . call_count == <NUM_LIT:1> <EOL> def test_content_length_reads_dont_call_close_callback ( self ) : <EOL> connection = mock . MagicMock ( ) <EOL> headers = { b'<STR_LIT>' : [ b'<STR_LIT>' ] } <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , connection ) <EOL> d . _buffer = BytesIO ( b'<STR_LIT>' ) <EOL> r . read ( ) <EOL> assert r . _sock is None <EOL> assert connection . close . call_count == <NUM_LIT:0> <EOL> def test_chunked_reads_dont_call_close_callback ( self ) : <EOL> connection = mock . MagicMock ( ) <EOL> headers = { b'<STR_LIT>' : [ b'<STR_LIT>' ] } <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , connection ) <EOL> data = ( <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> ) <EOL> d . _buffer = BytesIO ( data ) <EOL> list ( r . read_chunked ( ) ) <EOL> assert r . _sock is None <EOL> assert connection . close . call_count == <NUM_LIT:0> <EOL> def test_closing_chunked_reads_dont_call_close_callback ( self ) : <EOL> connection = mock . MagicMock ( ) <EOL> headers = { <EOL> b'<STR_LIT>' : [ b'<STR_LIT>' ] , b'<STR_LIT>' : [ b'<STR_LIT>' ] <EOL> } <EOL> d = DummySocket ( ) <EOL> r = HTTP11Response ( <NUM_LIT:200> , '<STR_LIT:OK>' , headers , d , connection ) <EOL> data = ( <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> b'<STR_LIT>' <EOL> ) <EOL> d . _buffer = BytesIO ( data ) <EOL> list ( r . read_chunked ( ) ) <EOL> assert r . _sock is None <EOL> assert connection . close . call_count == <NUM_LIT:1> <EOL> class DummySocket ( object ) : <EOL> def __init__ ( self ) : <EOL> self . queue = [ ] <EOL> self . _buffer = BytesIO ( ) <EOL> self . _read_counter = <NUM_LIT:0> <EOL> self . can_read = False <EOL> @ property <EOL> def buffer ( self ) : <EOL> return memoryview ( self . _buffer . getvalue ( ) [ self . _read_counter : ] ) <EOL> def advance_buffer ( self , amt ) : <EOL> self . _read_counter += amt <EOL> self . _buffer . read ( amt ) <EOL> def send ( self , data ) : <EOL> if not isinstance ( data , bytes ) : <EOL> raise TypeError ( ) <EOL> self . queue . append ( data ) <EOL> def recv ( self , l ) : <EOL> data = self . _buffer . read ( l ) <EOL> self . _read_counter += len ( data ) <EOL> return memoryview ( data ) <EOL> def close ( self ) : <EOL> pass <EOL> def readline ( self ) : <EOL> line = self . _buffer . readline ( ) <EOL> self . _read_counter += len ( line ) <EOL> return memoryview ( line ) <EOL> def fill ( self ) : <EOL> pass <EOL> class DummyFile ( object ) : <EOL> def __init__ ( self , data ) : <EOL> self . buffer = BytesIO ( data ) <EOL> def read ( self , amt = None ) : <EOL> return self . buffer . read ( amt ) <EOL> def fileno ( self ) : <EOL> return - <NUM_LIT:1> <EOL> def readline ( self ) : <EOL> self . buffer . readline ( ) <EOL> def __iter__ ( self ) : <EOL> return self . buffer . __iter__ ( ) </s>
