<s> """<STR_LIT>""" <EOL> from urlparse import urlparse , urlunsplit , clear_cache <EOL> import random , urllib , cgi <EOL> from twisted . python . compat import set <EOL> from twisted . python . failure import Failure <EOL> from twisted . trial import unittest <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . web import http , http_headers <EOL> from twisted . web . http import PotentialDataLoss , _DataLoss <EOL> from twisted . web . http import _IdentityTransferDecoder <EOL> from twisted . protocols import loopback <EOL> from twisted . internet . task import Clock <EOL> from twisted . internet . error import ConnectionLost <EOL> from twisted . test . proto_helpers import StringTransport <EOL> from twisted . test . test_internet import DummyProducer <EOL> from twisted . web . test . test_web import DummyChannel <EOL> class DateTimeTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testRoundtrip ( self ) : <EOL> for i in range ( <NUM_LIT> ) : <EOL> time = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> timestr = http . datetimeToString ( time ) <EOL> time2 = http . stringToDatetime ( timestr ) <EOL> self . assertEquals ( time , time2 ) <EOL> class DummyHTTPHandler ( http . Request ) : <EOL> def process ( self ) : <EOL> self . content . seek ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> data = self . content . read ( ) <EOL> length = self . getHeader ( '<STR_LIT>' ) <EOL> request = "<STR_LIT>" + str ( length ) + "<STR_LIT:\n>" + data + "<STR_LIT>" <EOL> self . setResponseCode ( <NUM_LIT:200> ) <EOL> self . setHeader ( "<STR_LIT>" , self . uri ) <EOL> self . setHeader ( "<STR_LIT>" , self . method ) <EOL> self . setHeader ( "<STR_LIT>" , self . clientproto ) <EOL> self . setHeader ( "<STR_LIT>" , len ( request ) ) <EOL> self . write ( request ) <EOL> self . finish ( ) <EOL> class LoopbackHTTPClient ( http . HTTPClient ) : <EOL> def connectionMade ( self ) : <EOL> self . sendCommand ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . sendHeader ( "<STR_LIT>" , <NUM_LIT:10> ) <EOL> self . endHeaders ( ) <EOL> self . transport . write ( "<STR_LIT>" ) <EOL> class ResponseTestMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def assertResponseEquals ( self , responses , expected ) : <EOL> """<STR_LIT>""" <EOL> for response in expected : <EOL> expectedHeaders , expectedContent = response [ : - <NUM_LIT:1> ] , response [ - <NUM_LIT:1> ] <EOL> headers , rest = responses . split ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> headers = headers . splitlines ( ) <EOL> self . assertEqual ( set ( headers ) , set ( expectedHeaders ) ) <EOL> content = rest [ : len ( expectedContent ) ] <EOL> responses = rest [ len ( expectedContent ) : ] <EOL> self . assertEqual ( content , expectedContent ) <EOL> class HTTP1_0TestCase ( unittest . TestCase , ResponseTestMixin ) : <EOL> requests = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" ) <EOL> expected_response = [ <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ] <EOL> def test_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> b = StringTransport ( ) <EOL> a = http . HTTPChannel ( ) <EOL> a . requestFactory = DummyHTTPHandler <EOL> a . makeConnection ( b ) <EOL> for byte in self . requests : <EOL> a . dataReceived ( byte ) <EOL> a . connectionLost ( IOError ( "<STR_LIT>" ) ) <EOL> value = b . value ( ) <EOL> self . assertResponseEquals ( value , self . expected_response ) <EOL> def test_requestBodyTimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> clock = Clock ( ) <EOL> transport = StringTransport ( ) <EOL> protocol = http . HTTPChannel ( ) <EOL> protocol . timeOut = <NUM_LIT:100> <EOL> protocol . callLater = clock . callLater <EOL> protocol . makeConnection ( transport ) <EOL> protocol . dataReceived ( '<STR_LIT>' ) <EOL> clock . advance ( <NUM_LIT> ) <EOL> self . assertFalse ( transport . disconnecting ) <EOL> protocol . dataReceived ( '<STR_LIT:x>' ) <EOL> clock . advance ( <NUM_LIT> ) <EOL> self . assertFalse ( transport . disconnecting ) <EOL> protocol . dataReceived ( '<STR_LIT:x>' ) <EOL> self . assertEqual ( len ( protocol . requests ) , <NUM_LIT:1> ) <EOL> class HTTP1_1TestCase ( HTTP1_0TestCase ) : <EOL> requests = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" ) <EOL> expected_response = [ <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ] <EOL> class HTTP1_1_close_TestCase ( HTTP1_0TestCase ) : <EOL> requests = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:\r\n>" ) <EOL> expected_response = [ <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ] <EOL> class HTTP0_9TestCase ( HTTP1_0TestCase ) : <EOL> requests = ( <EOL> "<STR_LIT>" ) <EOL> expected_response = "<STR_LIT>" <EOL> def assertResponseEquals ( self , response , expectedResponse ) : <EOL> self . assertEquals ( response , expectedResponse ) <EOL> class HTTPLoopbackTestCase ( unittest . TestCase ) : <EOL> expectedHeaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:GET>' , <EOL> '<STR_LIT:version>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> numHeaders = <NUM_LIT:0> <EOL> gotStatus = <NUM_LIT:0> <EOL> gotResponse = <NUM_LIT:0> <EOL> gotEndHeaders = <NUM_LIT:0> <EOL> def _handleStatus ( self , version , status , message ) : <EOL> self . gotStatus = <NUM_LIT:1> <EOL> self . assertEquals ( version , "<STR_LIT>" ) <EOL> self . assertEquals ( status , "<STR_LIT>" ) <EOL> def _handleResponse ( self , data ) : <EOL> self . gotResponse = <NUM_LIT:1> <EOL> self . assertEquals ( data , "<STR_LIT>" ) <EOL> def _handleHeader ( self , key , value ) : <EOL> self . numHeaders = self . numHeaders + <NUM_LIT:1> <EOL> self . assertEquals ( self . expectedHeaders [ key . lower ( ) ] , value ) <EOL> def _handleEndHeaders ( self ) : <EOL> self . gotEndHeaders = <NUM_LIT:1> <EOL> self . assertEquals ( self . numHeaders , <NUM_LIT:4> ) <EOL> def testLoopback ( self ) : <EOL> server = http . HTTPChannel ( ) <EOL> server . requestFactory = DummyHTTPHandler <EOL> client = LoopbackHTTPClient ( ) <EOL> client . handleResponse = self . _handleResponse <EOL> client . handleHeader = self . _handleHeader <EOL> client . handleEndHeaders = self . _handleEndHeaders <EOL> client . handleStatus = self . _handleStatus <EOL> d = loopback . loopbackAsync ( server , client ) <EOL> d . addCallback ( self . _cbTestLoopback ) <EOL> return d <EOL> def _cbTestLoopback ( self , ignored ) : <EOL> if not ( self . gotStatus and self . gotResponse and self . gotEndHeaders ) : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> % [ self . gotStatus , self . gotResponse , self . gotEndHeaders ] ) <EOL> del self . gotEndHeaders <EOL> del self . gotResponse <EOL> del self . gotStatus <EOL> del self . numHeaders <EOL> def _prequest ( ** headers ) : <EOL> """<STR_LIT>""" <EOL> request = http . Request ( DummyChannel ( ) , None ) <EOL> for k , v in headers . iteritems ( ) : <EOL> request . requestHeaders . setRawHeaders ( k , v ) <EOL> return request <EOL> class PersistenceTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> ptests = [ <EOL> ( _prequest ( ) , "<STR_LIT>" , <NUM_LIT:0> , { '<STR_LIT>' : None } ) , <EOL> ( _prequest ( connection = [ "<STR_LIT>" ] ) , "<STR_LIT>" , <NUM_LIT:0> , { '<STR_LIT>' : [ '<STR_LIT>' ] } ) , <EOL> ( _prequest ( ) , "<STR_LIT>" , <NUM_LIT:1> , { '<STR_LIT>' : None } ) , <EOL> ( _prequest ( ) , "<STR_LIT>" , <NUM_LIT:0> , { '<STR_LIT>' : None } ) , <EOL> ] <EOL> def testAlgorithm ( self ) : <EOL> c = http . HTTPChannel ( ) <EOL> for req , version , correctResult , resultHeaders in self . ptests : <EOL> result = c . checkPersistence ( req , version ) <EOL> self . assertEquals ( result , correctResult ) <EOL> for header in resultHeaders . keys ( ) : <EOL> self . assertEquals ( req . responseHeaders . getRawHeaders ( header , None ) , resultHeaders [ header ] ) <EOL> class IdentityTransferEncodingTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . data = [ ] <EOL> self . finish = [ ] <EOL> self . contentLength = <NUM_LIT:10> <EOL> self . decoder = _IdentityTransferDecoder ( <EOL> self . contentLength , self . data . append , self . finish . append ) <EOL> def test_exactAmountReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> self . decoder . dataReceived ( '<STR_LIT:x>' * self . contentLength ) <EOL> self . assertEqual ( self . data , [ '<STR_LIT:x>' * self . contentLength ] ) <EOL> self . assertEqual ( self . finish , [ '<STR_LIT>' ] ) <EOL> def test_shortStrings ( self ) : <EOL> """<STR_LIT>""" <EOL> self . decoder . dataReceived ( '<STR_LIT:x>' ) <EOL> self . assertEqual ( self . data , [ '<STR_LIT:x>' ] ) <EOL> self . assertEqual ( self . finish , [ ] ) <EOL> self . decoder . dataReceived ( '<STR_LIT:y>' * ( self . contentLength - <NUM_LIT:1> ) ) <EOL> self . assertEqual ( self . data , [ '<STR_LIT:x>' , '<STR_LIT:y>' * ( self . contentLength - <NUM_LIT:1> ) ] ) <EOL> self . assertEqual ( self . finish , [ '<STR_LIT>' ] ) <EOL> def test_longString ( self ) : <EOL> """<STR_LIT>""" <EOL> self . decoder . dataReceived ( '<STR_LIT:x>' * self . contentLength + '<STR_LIT:y>' ) <EOL> self . assertEqual ( self . data , [ '<STR_LIT:x>' * self . contentLength ] ) <EOL> self . assertEqual ( self . finish , [ '<STR_LIT:y>' ] ) <EOL> def test_rejectDataAfterFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> failures = [ ] <EOL> def finish ( bytes ) : <EOL> try : <EOL> decoder . dataReceived ( '<STR_LIT:foo>' ) <EOL> except : <EOL> failures . append ( Failure ( ) ) <EOL> decoder = _IdentityTransferDecoder ( <NUM_LIT:5> , self . data . append , finish ) <EOL> decoder . dataReceived ( '<STR_LIT:x>' * <NUM_LIT:4> ) <EOL> self . assertEqual ( failures , [ ] ) <EOL> decoder . dataReceived ( '<STR_LIT:y>' ) <EOL> failures [ <NUM_LIT:0> ] . trap ( RuntimeError ) <EOL> self . assertEqual ( <EOL> str ( failures [ <NUM_LIT:0> ] . value ) , <EOL> "<STR_LIT>" ) <EOL> def test_unknownContentLength ( self ) : <EOL> """<STR_LIT>""" <EOL> data = [ ] <EOL> finish = [ ] <EOL> decoder = _IdentityTransferDecoder ( None , data . append , finish . append ) <EOL> decoder . dataReceived ( '<STR_LIT:x>' ) <EOL> self . assertEqual ( data , [ '<STR_LIT:x>' ] ) <EOL> decoder . dataReceived ( '<STR_LIT:y>' ) <EOL> self . assertEqual ( data , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) <EOL> self . assertEqual ( finish , [ ] ) <EOL> def _verifyCallbacksUnreferenced ( self , decoder ) : <EOL> """<STR_LIT>""" <EOL> self . assertIdentical ( decoder . dataCallback , None ) <EOL> self . assertIdentical ( decoder . finishCallback , None ) <EOL> def test_earlyConnectionLose ( self ) : <EOL> """<STR_LIT>""" <EOL> self . decoder . dataReceived ( '<STR_LIT:x>' * ( self . contentLength - <NUM_LIT:1> ) ) <EOL> self . assertRaises ( _DataLoss , self . decoder . noMoreData ) <EOL> self . _verifyCallbacksUnreferenced ( self . decoder ) <EOL> def test_unknownContentLengthConnectionLose ( self ) : <EOL> """<STR_LIT>""" <EOL> body = [ ] <EOL> finished = [ ] <EOL> decoder = _IdentityTransferDecoder ( None , body . append , finished . append ) <EOL> self . assertRaises ( PotentialDataLoss , decoder . noMoreData ) <EOL> self . assertEqual ( body , [ ] ) <EOL> self . assertEqual ( finished , [ '<STR_LIT>' ] ) <EOL> self . _verifyCallbacksUnreferenced ( decoder ) <EOL> def test_finishedConnectionLose ( self ) : <EOL> """<STR_LIT>""" <EOL> self . decoder . dataReceived ( '<STR_LIT:x>' * self . contentLength ) <EOL> self . decoder . noMoreData ( ) <EOL> self . _verifyCallbacksUnreferenced ( self . decoder ) <EOL> class ChunkedTransferEncodingTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_decoding ( self ) : <EOL> """<STR_LIT>""" <EOL> L = [ ] <EOL> p = http . _ChunkedTransferDecoder ( L . append , None ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( L , [ '<STR_LIT:abc>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_short ( self ) : <EOL> """<STR_LIT>""" <EOL> L = [ ] <EOL> finished = [ ] <EOL> p = http . _ChunkedTransferDecoder ( L . append , finished . append ) <EOL> for s in '<STR_LIT>' : <EOL> p . dataReceived ( s ) <EOL> self . assertEqual ( L , [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' , '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' , '<STR_LIT:4>' , '<STR_LIT:5>' ] ) <EOL> self . assertEqual ( finished , [ '<STR_LIT>' ] ) <EOL> def test_newlines ( self ) : <EOL> """<STR_LIT>""" <EOL> L = [ ] <EOL> p = http . _ChunkedTransferDecoder ( L . append , None ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( L , [ '<STR_LIT:\r\n>' ] ) <EOL> def test_extensions ( self ) : <EOL> """<STR_LIT>""" <EOL> L = [ ] <EOL> p = http . _ChunkedTransferDecoder ( L . append , None ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( L , [ '<STR_LIT:abc>' ] ) <EOL> def test_finish ( self ) : <EOL> """<STR_LIT>""" <EOL> finished = [ ] <EOL> p = http . _ChunkedTransferDecoder ( None , finished . append ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( finished , [ '<STR_LIT>' ] ) <EOL> def test_extra ( self ) : <EOL> """<STR_LIT>""" <EOL> finished = [ ] <EOL> p = http . _ChunkedTransferDecoder ( None , finished . append ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( finished , [ '<STR_LIT:hello>' ] ) <EOL> def test_afterFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> p = http . _ChunkedTransferDecoder ( None , lambda bytes : None ) <EOL> p . dataReceived ( '<STR_LIT>' ) <EOL> self . assertRaises ( RuntimeError , p . dataReceived , '<STR_LIT:hello>' ) <EOL> def test_earlyConnectionLose ( self ) : <EOL> """<STR_LIT>""" <EOL> parser = http . _ChunkedTransferDecoder ( None , lambda bytes : None ) <EOL> parser . dataReceived ( '<STR_LIT>' ) <EOL> exc = self . assertRaises ( _DataLoss , parser . noMoreData ) <EOL> self . assertEqual ( <EOL> str ( exc ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_finishedConnectionLose ( self ) : <EOL> """<STR_LIT>""" <EOL> parser = http . _ChunkedTransferDecoder ( None , lambda bytes : None ) <EOL> parser . dataReceived ( '<STR_LIT>' ) <EOL> parser . noMoreData ( ) <EOL> def test_reentrantFinishedNoMoreData ( self ) : <EOL> """<STR_LIT>""" <EOL> errors = [ ] <EOL> successes = [ ] <EOL> def finished ( extra ) : <EOL> try : <EOL> parser . noMoreData ( ) <EOL> except : <EOL> errors . append ( Failure ( ) ) <EOL> else : <EOL> successes . append ( True ) <EOL> parser = http . _ChunkedTransferDecoder ( None , finished ) <EOL> parser . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( errors , [ ] ) <EOL> self . assertEqual ( successes , [ True ] ) <EOL> class ChunkingTestCase ( unittest . TestCase ) : <EOL> strings = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> def testChunks ( self ) : <EOL> for s in self . strings : <EOL> self . assertEquals ( ( s , '<STR_LIT>' ) , http . fromChunk ( '<STR_LIT>' . join ( http . toChunk ( s ) ) ) ) <EOL> self . assertRaises ( ValueError , http . fromChunk , '<STR_LIT>' ) <EOL> def testConcatenatedChunks ( self ) : <EOL> chunked = '<STR_LIT>' . join ( [ '<STR_LIT>' . join ( http . toChunk ( t ) ) for t in self . strings ] ) <EOL> result = [ ] <EOL> buffer = "<STR_LIT>" <EOL> for c in chunked : <EOL> buffer = buffer + c <EOL> try : <EOL> data , buffer = http . fromChunk ( buffer ) <EOL> result . append ( data ) <EOL> except ValueError : <EOL> pass <EOL> self . assertEquals ( result , self . strings ) <EOL> class ParsingTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def runRequest ( self , httpRequest , requestClass , success = <NUM_LIT:1> ) : <EOL> httpRequest = httpRequest . replace ( "<STR_LIT:\n>" , "<STR_LIT:\r\n>" ) <EOL> b = StringTransport ( ) <EOL> a = http . HTTPChannel ( ) <EOL> a . requestFactory = requestClass <EOL> a . makeConnection ( b ) <EOL> for byte in httpRequest : <EOL> if a . transport . disconnecting : <EOL> break <EOL> a . dataReceived ( byte ) <EOL> a . connectionLost ( IOError ( "<STR_LIT>" ) ) <EOL> if success : <EOL> self . assertEquals ( self . didRequest , <NUM_LIT:1> ) <EOL> del self . didRequest <EOL> else : <EOL> self . assert_ ( not hasattr ( self , "<STR_LIT>" ) ) <EOL> return a <EOL> def test_basicAuth ( self ) : <EOL> """<STR_LIT>""" <EOL> testcase = self <EOL> class Request ( http . Request ) : <EOL> l = [ ] <EOL> def process ( self ) : <EOL> testcase . assertEquals ( self . getUser ( ) , self . l [ <NUM_LIT:0> ] ) <EOL> testcase . assertEquals ( self . getPassword ( ) , self . l [ <NUM_LIT:1> ] ) <EOL> for u , p in [ ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , ( "<STR_LIT:hello>" , "<STR_LIT>" ) ] : <EOL> Request . l [ : ] = [ u , p ] <EOL> s = "<STR_LIT>" % ( u , p ) <EOL> f = "<STR_LIT>" % ( s . encode ( "<STR_LIT>" ) . strip ( ) , ) <EOL> self . runRequest ( f , Request , <NUM_LIT:0> ) <EOL> def test_headers ( self ) : <EOL> """<STR_LIT>""" <EOL> processed = [ ] <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> processed . append ( self ) <EOL> self . finish ( ) <EOL> requestLines = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> self . runRequest ( '<STR_LIT:\n>' . join ( requestLines ) , MyRequest , <NUM_LIT:0> ) <EOL> [ request ] = processed <EOL> self . assertEquals ( <EOL> request . requestHeaders . getRawHeaders ( '<STR_LIT:foo>' ) , [ '<STR_LIT:bar>' ] ) <EOL> self . assertEquals ( <EOL> request . requestHeaders . getRawHeaders ( '<STR_LIT>' ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_tooManyHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> processed = [ ] <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> processed . append ( self ) <EOL> requestLines = [ "<STR_LIT>" ] <EOL> for i in range ( http . HTTPChannel . maxHeaders + <NUM_LIT:2> ) : <EOL> requestLines . append ( "<STR_LIT>" % ( i , ) ) <EOL> requestLines . extend ( [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> channel = self . runRequest ( "<STR_LIT:\n>" . join ( requestLines ) , MyRequest , <NUM_LIT:0> ) <EOL> self . assertEqual ( processed , [ ] ) <EOL> self . assertEqual ( <EOL> channel . transport . value ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_headerLimitPerRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> processed = [ ] <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> processed . append ( self ) <EOL> self . finish ( ) <EOL> self . patch ( http . HTTPChannel , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> requestLines = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> channel = self . runRequest ( "<STR_LIT:\n>" . join ( requestLines ) , MyRequest , <NUM_LIT:0> ) <EOL> [ first , second ] = processed <EOL> self . assertEqual ( first . getHeader ( '<STR_LIT:foo>' ) , '<STR_LIT:bar>' ) <EOL> self . assertEqual ( second . getHeader ( '<STR_LIT:bar>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> channel . transport . value ( ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:\r\n>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:\r\n>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:\r\n>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:\r\n>' ) <EOL> def testCookies ( self ) : <EOL> """<STR_LIT>""" <EOL> httpRequest = '''<STR_LIT>''' <EOL> testcase = self <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> testcase . assertEquals ( self . getCookie ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> testcase . assertEquals ( self . getCookie ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> testcase . assertEquals ( self . getCookie ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> testcase . didRequest = <NUM_LIT:1> <EOL> self . finish ( ) <EOL> self . runRequest ( httpRequest , MyRequest ) <EOL> def testGET ( self ) : <EOL> httpRequest = '''<STR_LIT>''' <EOL> testcase = self <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> testcase . assertEquals ( self . method , "<STR_LIT:GET>" ) <EOL> testcase . assertEquals ( self . args [ "<STR_LIT:key>" ] , [ "<STR_LIT:value>" ] ) <EOL> testcase . assertEquals ( self . args [ "<STR_LIT>" ] , [ "<STR_LIT>" ] ) <EOL> testcase . assertEquals ( self . args [ "<STR_LIT>" ] , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> testcase . didRequest = <NUM_LIT:1> <EOL> self . finish ( ) <EOL> self . runRequest ( httpRequest , MyRequest ) <EOL> def test_extraQuestionMark ( self ) : <EOL> """<STR_LIT>""" <EOL> httpRequest = '<STR_LIT>' <EOL> testcase = self <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> testcase . assertEqual ( self . method , '<STR_LIT:GET>' ) <EOL> testcase . assertEqual ( self . path , '<STR_LIT>' ) <EOL> testcase . assertEqual ( self . args [ '<STR_LIT:bar>' ] , [ '<STR_LIT:?>' ] ) <EOL> testcase . assertEqual ( self . args [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) <EOL> testcase . didRequest = <NUM_LIT:1> <EOL> self . finish ( ) <EOL> self . runRequest ( httpRequest , MyRequest ) <EOL> def test_formPOSTRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> query = '<STR_LIT>' <EOL> httpRequest = '''<STR_LIT>''' % ( len ( query ) , query ) <EOL> testcase = self <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> testcase . assertEquals ( self . method , "<STR_LIT:POST>" ) <EOL> testcase . assertEquals ( self . args [ "<STR_LIT:key>" ] , [ "<STR_LIT:value>" ] ) <EOL> testcase . assertEquals ( self . args [ "<STR_LIT>" ] , [ "<STR_LIT>" ] ) <EOL> testcase . assertEquals ( self . args [ "<STR_LIT>" ] , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> testcase . assertEquals ( self . content . read ( ) , query ) <EOL> testcase . didRequest = <NUM_LIT:1> <EOL> self . finish ( ) <EOL> self . runRequest ( httpRequest , MyRequest ) <EOL> def testMissingContentDisposition ( self ) : <EOL> req = '''<STR_LIT>''' <EOL> self . runRequest ( req , http . Request , success = False ) <EOL> def test_chunkedEncoding ( self ) : <EOL> """<STR_LIT>""" <EOL> httpRequest = '''<STR_LIT>''' <EOL> testcase = self <EOL> class MyRequest ( http . Request ) : <EOL> def process ( self ) : <EOL> testcase . assertIsInstance ( self . content . fileno ( ) , int ) <EOL> testcase . assertEqual ( self . method , '<STR_LIT:GET>' ) <EOL> testcase . assertEqual ( self . path , '<STR_LIT:/>' ) <EOL> content = self . content . read ( ) <EOL> testcase . assertEqual ( content , '<STR_LIT>' ) <EOL> testcase . assertIdentical ( self . channel . _transferDecoder , None ) <EOL> testcase . didRequest = <NUM_LIT:1> <EOL> self . finish ( ) <EOL> self . runRequest ( httpRequest , MyRequest ) <EOL> class QueryArgumentsTestCase ( unittest . TestCase ) : <EOL> def testUnquote ( self ) : <EOL> try : <EOL> from twisted . protocols import _c_urlarg <EOL> except ImportError : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> self . failUnlessEqual ( urllib . unquote ( "<STR_LIT>" ) , <EOL> _c_urlarg . unquote ( "<STR_LIT>" ) ) <EOL> self . failUnlessEqual ( urllib . unquote ( "<STR_LIT>" ) , <EOL> _c_urlarg . unquote ( "<STR_LIT>" ) ) <EOL> self . failUnlessEqual ( urllib . unquote ( "<STR_LIT>" ) , <EOL> _c_urlarg . unquote ( "<STR_LIT>" ) ) <EOL> self . failUnlessEqual ( urllib . unquote ( "<STR_LIT>" ) , _c_urlarg . unquote ( "<STR_LIT>" ) ) <EOL> def testParseqs ( self ) : <EOL> self . failUnlessEqual ( cgi . parse_qs ( "<STR_LIT>" ) , <EOL> http . parse_qs ( "<STR_LIT>" ) ) <EOL> self . failUnlessRaises ( ValueError , http . parse_qs , "<STR_LIT>" , <EOL> strict_parsing = <NUM_LIT:1> ) <EOL> self . failUnlessEqual ( cgi . parse_qs ( "<STR_LIT>" , keep_blank_values = <NUM_LIT:1> ) , <EOL> http . parse_qs ( "<STR_LIT>" , keep_blank_values = <NUM_LIT:1> ) ) <EOL> self . failUnlessEqual ( cgi . parse_qs ( "<STR_LIT>" ) , <EOL> http . parse_qs ( "<STR_LIT>" ) ) <EOL> def test_urlparse ( self ) : <EOL> """<STR_LIT>""" <EOL> def urls ( ) : <EOL> for scheme in ( '<STR_LIT:http>' , '<STR_LIT>' ) : <EOL> for host in ( '<STR_LIT>' , ) : <EOL> for port in ( None , <NUM_LIT:100> ) : <EOL> for path in ( '<STR_LIT>' , '<STR_LIT:path>' ) : <EOL> if port is not None : <EOL> host = host + '<STR_LIT::>' + str ( port ) <EOL> yield urlunsplit ( ( scheme , host , path , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def assertSameParsing ( url , decode ) : <EOL> """<STR_LIT>""" <EOL> urlToStandardImplementation = url <EOL> if decode : <EOL> urlToStandardImplementation = url . decode ( '<STR_LIT:ascii>' ) <EOL> standardResult = urlparse ( urlToStandardImplementation ) <EOL> scheme , netloc , path , params , query , fragment = http . urlparse ( url ) <EOL> self . assertEqual ( <EOL> ( scheme , netloc , path , params , query , fragment ) , <EOL> standardResult ) <EOL> self . assertTrue ( isinstance ( scheme , str ) ) <EOL> self . assertTrue ( isinstance ( netloc , str ) ) <EOL> self . assertTrue ( isinstance ( path , str ) ) <EOL> self . assertTrue ( isinstance ( params , str ) ) <EOL> self . assertTrue ( isinstance ( query , str ) ) <EOL> self . assertTrue ( isinstance ( fragment , str ) ) <EOL> clear_cache ( ) <EOL> for url in urls ( ) : <EOL> assertSameParsing ( url , True ) <EOL> assertSameParsing ( url , False ) <EOL> clear_cache ( ) <EOL> for url in urls ( ) : <EOL> assertSameParsing ( url , False ) <EOL> assertSameParsing ( url , True ) <EOL> for url in urls ( ) : <EOL> clear_cache ( ) <EOL> assertSameParsing ( url , True ) <EOL> clear_cache ( ) <EOL> assertSameParsing ( url , False ) <EOL> def test_urlparseRejectsUnicode ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( TypeError , http . urlparse , u'<STR_LIT>' ) <EOL> def testEscchar ( self ) : <EOL> try : <EOL> from twisted . protocols import _c_urlarg <EOL> except ImportError : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> self . failUnlessEqual ( "<STR_LIT>" , <EOL> _c_urlarg . unquote ( "<STR_LIT>" , "<STR_LIT:+>" ) ) <EOL> class ClientDriver ( http . HTTPClient ) : <EOL> def handleStatus ( self , version , status , message ) : <EOL> self . version = version <EOL> self . status = status <EOL> self . message = message <EOL> class ClientStatusParsing ( unittest . TestCase ) : <EOL> def testBaseline ( self ) : <EOL> c = ClientDriver ( ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . version , '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . status , '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . message , '<STR_LIT:foo>' ) <EOL> def testNoMessage ( self ) : <EOL> c = ClientDriver ( ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . version , '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . status , '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . message , '<STR_LIT>' ) <EOL> def testNoMessage_trailingSpace ( self ) : <EOL> c = ClientDriver ( ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . version , '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . status , '<STR_LIT>' ) <EOL> self . failUnlessEqual ( c . message , '<STR_LIT>' ) <EOL> class RequestTests ( unittest . TestCase , ResponseTestMixin ) : <EOL> """<STR_LIT>""" <EOL> def _compatHeadersTest ( self , oldName , newName ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> getattr ( req , newName ) . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( getattr ( req , oldName ) [ "<STR_LIT:test>" ] , "<STR_LIT>" ) <EOL> setattr ( req , oldName , { "<STR_LIT:foo>" : "<STR_LIT:bar>" } ) <EOL> self . assertEqual ( <EOL> list ( getattr ( req , newName ) . getAllRawHeaders ( ) ) , <EOL> [ ( "<STR_LIT>" , [ "<STR_LIT:bar>" ] ) ] ) <EOL> setattr ( req , newName , http_headers . Headers ( ) ) <EOL> self . assertEqual ( getattr ( req , oldName ) , { } ) <EOL> def test_received_headers ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _compatHeadersTest ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_headers ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _compatHeadersTest ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_getHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . requestHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" ] ) <EOL> self . assertEquals ( req . getHeader ( "<STR_LIT:test>" ) , "<STR_LIT>" ) <EOL> def test_getHeaderReceivedMultiples ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . requestHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( req . getHeader ( "<STR_LIT:test>" ) , "<STR_LIT>" ) <EOL> def test_getHeaderNotFound ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> self . assertEquals ( req . getHeader ( "<STR_LIT:test>" ) , None ) <EOL> def test_getAllHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . requestHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" ] ) <EOL> self . assertEquals ( req . getAllHeaders ( ) , { "<STR_LIT:test>" : "<STR_LIT>" } ) <EOL> def test_getAllHeadersNoHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> self . assertEquals ( req . getAllHeaders ( ) , { } ) <EOL> def test_getAllHeadersMultipleHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . requestHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( req . getAllHeaders ( ) , { "<STR_LIT:test>" : "<STR_LIT>" } ) <EOL> def test_setResponseCode ( self ) : <EOL> """<STR_LIT>""" <EOL> channel = DummyChannel ( ) <EOL> req = http . Request ( channel , None ) <EOL> req . setResponseCode ( <NUM_LIT> ) <EOL> req . write ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> channel . transport . written . getvalue ( ) . splitlines ( ) [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' % ( req . clientproto , ) ) <EOL> def test_setResponseCodeAndMessage ( self ) : <EOL> """<STR_LIT>""" <EOL> channel = DummyChannel ( ) <EOL> req = http . Request ( channel , None ) <EOL> req . setResponseCode ( <NUM_LIT> , "<STR_LIT>" ) <EOL> req . write ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> channel . transport . written . getvalue ( ) . splitlines ( ) [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' % ( req . clientproto , ) ) <EOL> def test_setResponseCodeAcceptsIntegers ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . setResponseCode ( <NUM_LIT:1> ) <EOL> req . setResponseCode ( <NUM_LIT:1> L ) <EOL> self . assertRaises ( TypeError , req . setResponseCode , "<STR_LIT:1>" ) <EOL> def test_setHost ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . setHost ( "<STR_LIT>" , <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> req . requestHeaders . getRawHeaders ( "<STR_LIT:host>" ) , [ "<STR_LIT>" ] ) <EOL> def test_setHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . setHeader ( "<STR_LIT:test>" , "<STR_LIT>" ) <EOL> self . assertEquals ( req . responseHeaders . getRawHeaders ( "<STR_LIT:test>" ) , [ "<STR_LIT>" ] ) <EOL> def test_firstWrite ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> trans = StringTransport ( ) <EOL> req . transport = trans <EOL> req . setResponseCode ( <NUM_LIT:200> ) <EOL> req . clientproto = "<STR_LIT>" <EOL> req . responseHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" ] ) <EOL> req . write ( '<STR_LIT>' ) <EOL> self . assertResponseEquals ( <EOL> trans . value ( ) , <EOL> [ ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ] ) <EOL> def test_firstWriteHTTP11Chunked ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> trans = StringTransport ( ) <EOL> req . transport = trans <EOL> req . setResponseCode ( <NUM_LIT:200> ) <EOL> req . clientproto = "<STR_LIT>" <EOL> req . responseHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" ] ) <EOL> req . write ( '<STR_LIT>' ) <EOL> req . write ( '<STR_LIT>' ) <EOL> self . assertResponseEquals ( <EOL> trans . value ( ) , <EOL> [ ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ] ) <EOL> def test_firstWriteLastModified ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> trans = StringTransport ( ) <EOL> req . transport = trans <EOL> req . setResponseCode ( <NUM_LIT:200> ) <EOL> req . clientproto = "<STR_LIT>" <EOL> req . lastModified = <NUM_LIT:0> <EOL> req . responseHeaders . setRawHeaders ( "<STR_LIT:test>" , [ "<STR_LIT>" ] ) <EOL> req . write ( '<STR_LIT>' ) <EOL> self . assertResponseEquals ( <EOL> trans . value ( ) , <EOL> [ ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) ] ) <EOL> def test_parseCookies ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . requestHeaders . setRawHeaders ( <EOL> "<STR_LIT>" , [ '<STR_LIT>' ] ) <EOL> req . parseCookies ( ) <EOL> self . assertEquals ( req . received_cookies , { "<STR_LIT:test>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : '<STR_LIT>' } ) <EOL> def test_parseCookiesMultipleHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . requestHeaders . setRawHeaders ( <EOL> "<STR_LIT>" , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> req . parseCookies ( ) <EOL> self . assertEquals ( req . received_cookies , { "<STR_LIT:test>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : '<STR_LIT>' } ) <EOL> def test_connectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . gotLength ( <NUM_LIT:10> ) <EOL> content = req . content <EOL> req . handleContentChunk ( "<STR_LIT:hello>" ) <EOL> req . connectionLost ( Failure ( ConnectionLost ( "<STR_LIT>" ) ) ) <EOL> self . assertTrue ( content . closed ) <EOL> self . assertIdentical ( req . channel , None ) <EOL> def test_registerProducerTwiceFails ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , None ) <EOL> req . registerProducer ( DummyProducer ( ) , True ) <EOL> self . assertRaises ( <EOL> ValueError , req . registerProducer , DummyProducer ( ) , True ) <EOL> def test_registerProducerWhenQueuedPausesPushProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , True ) <EOL> producer = DummyProducer ( ) <EOL> req . registerProducer ( producer , True ) <EOL> self . assertEquals ( [ '<STR_LIT>' ] , producer . events ) <EOL> def test_registerProducerWhenQueuedDoesntPausePullProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , True ) <EOL> producer = DummyProducer ( ) <EOL> req . registerProducer ( producer , False ) <EOL> self . assertEquals ( [ ] , producer . events ) <EOL> def test_registerProducerWhenQueuedDoesntRegisterPushProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertIdentical ( <EOL> None , getattr ( http . StringTransport , '<STR_LIT>' , None ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> req = http . Request ( DummyChannel ( ) , True ) <EOL> producer = DummyProducer ( ) <EOL> req . registerProducer ( producer , True ) <EOL> self . assertIsInstance ( req . transport , http . StringTransport ) <EOL> def test_registerProducerWhenQueuedDoesntRegisterPullProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertIdentical ( <EOL> None , getattr ( http . StringTransport , '<STR_LIT>' , None ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> req = http . Request ( DummyChannel ( ) , True ) <EOL> producer = DummyProducer ( ) <EOL> req . registerProducer ( producer , False ) <EOL> self . assertIsInstance ( req . transport , http . StringTransport ) <EOL> def test_registerProducerWhenNotQueuedRegistersPushProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , False ) <EOL> producer = DummyProducer ( ) <EOL> req . registerProducer ( producer , True ) <EOL> self . assertEquals ( [ ( producer , True ) ] , req . transport . producers ) <EOL> def test_registerProducerWhenNotQueuedRegistersPullProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> req = http . Request ( DummyChannel ( ) , False ) <EOL> producer = DummyProducer ( ) <EOL> req . registerProducer ( producer , False ) <EOL> self . assertEquals ( [ ( producer , False ) ] , req . transport . producers ) <EOL> def test_connectionLostNotification ( self ) : <EOL> """<STR_LIT>""" <EOL> d = DummyChannel ( ) <EOL> request = http . Request ( d , True ) <EOL> finished = request . notifyFinish ( ) <EOL> request . connectionLost ( Failure ( ConnectionLost ( "<STR_LIT>" ) ) ) <EOL> self . assertIdentical ( request . channel , None ) <EOL> return self . assertFailure ( finished , ConnectionLost ) <EOL> def test_finishNotification ( self ) : <EOL> """<STR_LIT>""" <EOL> request = http . Request ( DummyChannel ( ) , False ) <EOL> finished = request . notifyFinish ( ) <EOL> request . gotLength ( <NUM_LIT:1> ) <EOL> request . finish ( ) <EOL> return finished <EOL> def test_finishAfterConnectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> channel = DummyChannel ( ) <EOL> transport = channel . transport <EOL> req = http . Request ( channel , False ) <EOL> req . connectionLost ( Failure ( ConnectionLost ( "<STR_LIT>" ) ) ) <EOL> self . assertRaises ( RuntimeError , req . finish ) <EOL> class MultilineHeadersTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . handleHeaderCalled = False <EOL> self . handleEndHeadersCalled = False <EOL> expectedHeaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def ourHandleHeader ( self , key , val ) : <EOL> """<STR_LIT>""" <EOL> self . handleHeaderCalled = True <EOL> self . assertEquals ( val , self . expectedHeaders [ key ] ) <EOL> def ourHandleEndHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> self . handleEndHeadersCalled = True <EOL> def test_extractHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> c = ClientDriver ( ) <EOL> c . handleHeader = self . ourHandleHeader <EOL> c . handleEndHeaders = self . ourHandleEndHeaders <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . assertIdentical ( c . length , None ) <EOL> self . assertFalse ( self . handleHeaderCalled ) <EOL> self . assertFalse ( self . handleEndHeadersCalled ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . assertTrue ( self . handleHeaderCalled ) <EOL> self . assertTrue ( self . handleEndHeadersCalled ) <EOL> self . assertEquals ( c . length , <NUM_LIT:10> ) <EOL> def test_noHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> c = ClientDriver ( ) <EOL> c . handleHeader = self . ourHandleHeader <EOL> c . handleEndHeaders = self . ourHandleEndHeaders <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . assertFalse ( self . handleHeaderCalled ) <EOL> self . assertTrue ( self . handleEndHeadersCalled ) <EOL> self . assertEquals ( c . version , '<STR_LIT>' ) <EOL> self . assertEquals ( c . status , '<STR_LIT>' ) <EOL> def test_multilineHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> c = ClientDriver ( ) <EOL> c . handleHeader = self . ourHandleHeader <EOL> c . handleEndHeaders = self . ourHandleEndHeaders <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . assertFalse ( self . handleHeaderCalled ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . assertTrue ( self . handleHeaderCalled ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> c . lineReceived ( '<STR_LIT>' ) <EOL> self . assertTrue ( self . handleEndHeadersCalled ) <EOL> self . assertEquals ( c . version , '<STR_LIT>' ) <EOL> self . assertEquals ( c . status , '<STR_LIT>' ) <EOL> self . assertEquals ( c . length , <NUM_LIT:10> ) </s>
