<s> """<STR_LIT>""" <EOL> import cookielib <EOL> import zlib <EOL> from StringIO import StringIO <EOL> from zope . interface . verify import verifyObject <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . web import client , error , http_headers <EOL> from twisted . web . _newclient import RequestNotSent , RequestTransmissionFailed <EOL> from twisted . web . _newclient import ResponseNeverReceived , ResponseFailed <EOL> from twisted . web . _newclient import PotentialDataLoss <EOL> from twisted . internet import defer , task <EOL> from twisted . python . failure import Failure <EOL> from twisted . python . components import proxyForInterface <EOL> from twisted . test . proto_helpers import StringTransport , MemoryReactorClock <EOL> from twisted . internet . task import Clock <EOL> from twisted . internet . error import ConnectionRefusedError , ConnectionDone <EOL> from twisted . internet . error import ConnectionLost <EOL> from twisted . internet . protocol import Protocol , Factory <EOL> from twisted . internet . defer import Deferred , succeed , CancelledError <EOL> from twisted . internet . endpoints import TCP4ClientEndpoint , SSL4ClientEndpoint <EOL> from twisted . web . client import ( FileBodyProducer , Request , HTTPConnectionPool , <EOL> ResponseDone , _HTTP11ClientFactory , URI ) <EOL> from twisted . web . iweb import ( <EOL> UNKNOWN_LENGTH , IAgent , IBodyProducer , IResponse , IAgentEndpointFactory , <EOL> ) <EOL> from twisted . web . http_headers import Headers <EOL> from twisted . web . _newclient import HTTP11ClientProtocol , Response <EOL> from twisted . internet . interfaces import IOpenSSLClientConnectionCreator <EOL> from zope . interface . declarations import implementer <EOL> from twisted . web . iweb import IPolicyForHTTPS <EOL> from twisted . python . deprecate import getDeprecationWarningString <EOL> from twisted . python . versions import Version <EOL> from twisted . web . client import BrowserLikePolicyForHTTPS <EOL> from twisted . web . error import SchemeNotSupported <EOL> try : <EOL> from twisted . internet import ssl <EOL> from twisted . protocols . tls import TLSMemoryBIOFactory , TLSMemoryBIOProtocol <EOL> except ImportError : <EOL> ssl = None <EOL> else : <EOL> from twisted . internet . _sslverify import ClientTLSOptions , IOpenSSLTrustRoot <EOL> class StubHTTPProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . requests = [ ] <EOL> self . state = '<STR_LIT>' <EOL> def request ( self , request ) : <EOL> """<STR_LIT>""" <EOL> result = Deferred ( ) <EOL> self . requests . append ( ( request , result ) ) <EOL> return result <EOL> class FileConsumer ( object ) : <EOL> def __init__ ( self , outputFile ) : <EOL> self . outputFile = outputFile <EOL> def write ( self , bytes ) : <EOL> self . outputFile . write ( bytes ) <EOL> class FileBodyProducerTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _termination ( self ) : <EOL> """<STR_LIT>""" <EOL> return lambda : True <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _scheduled = [ ] <EOL> self . cooperator = task . Cooperator ( <EOL> self . _termination , self . _scheduled . append ) <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( verifyObject ( <EOL> IBodyProducer , FileBodyProducer ( StringIO ( "<STR_LIT>" ) ) ) ) <EOL> def test_unknownLength ( self ) : <EOL> """<STR_LIT>""" <EOL> class HasSeek ( object ) : <EOL> def seek ( self , offset , whence ) : <EOL> pass <EOL> class HasTell ( object ) : <EOL> def tell ( self ) : <EOL> pass <EOL> producer = FileBodyProducer ( HasSeek ( ) ) <EOL> self . assertEqual ( UNKNOWN_LENGTH , producer . length ) <EOL> producer = FileBodyProducer ( HasTell ( ) ) <EOL> self . assertEqual ( UNKNOWN_LENGTH , producer . length ) <EOL> def test_knownLength ( self ) : <EOL> """<STR_LIT>""" <EOL> inputBytes = "<STR_LIT>" <EOL> inputFile = StringIO ( inputBytes ) <EOL> inputFile . seek ( <NUM_LIT:5> ) <EOL> producer = FileBodyProducer ( inputFile ) <EOL> self . assertEqual ( len ( inputBytes ) - <NUM_LIT:5> , producer . length ) <EOL> self . assertEqual ( inputFile . tell ( ) , <NUM_LIT:5> ) <EOL> def test_defaultCooperator ( self ) : <EOL> """<STR_LIT>""" <EOL> producer = FileBodyProducer ( StringIO ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( task . cooperate , producer . _cooperate ) <EOL> def test_startProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedResult = "<STR_LIT>" <EOL> readSize = <NUM_LIT:3> <EOL> output = StringIO ( ) <EOL> consumer = FileConsumer ( output ) <EOL> producer = FileBodyProducer ( <EOL> StringIO ( expectedResult ) , self . cooperator , readSize ) <EOL> complete = producer . startProducing ( consumer ) <EOL> for i in range ( len ( expectedResult ) // readSize + <NUM_LIT:1> ) : <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertEqual ( [ ] , self . _scheduled ) <EOL> self . assertEqual ( expectedResult , output . getvalue ( ) ) <EOL> self . assertEqual ( None , self . successResultOf ( complete ) ) <EOL> def test_inputClosedAtEOF ( self ) : <EOL> """<STR_LIT>""" <EOL> readSize = <NUM_LIT:4> <EOL> inputBytes = "<STR_LIT>" <EOL> inputFile = StringIO ( inputBytes ) <EOL> producer = FileBodyProducer ( inputFile , self . cooperator , readSize ) <EOL> consumer = FileConsumer ( StringIO ( ) ) <EOL> producer . startProducing ( consumer ) <EOL> for i in range ( len ( inputBytes ) // readSize + <NUM_LIT:2> ) : <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertTrue ( inputFile . closed ) <EOL> def test_failedReadWhileProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> class BrokenFile ( object ) : <EOL> def read ( self , count ) : <EOL> raise IOError ( "<STR_LIT>" ) <EOL> producer = FileBodyProducer ( BrokenFile ( ) , self . cooperator ) <EOL> complete = producer . startProducing ( FileConsumer ( StringIO ( ) ) ) <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . failureResultOf ( complete ) . trap ( IOError ) <EOL> def test_stopProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedResult = "<STR_LIT>" <EOL> readSize = <NUM_LIT:3> <EOL> output = StringIO ( ) <EOL> consumer = FileConsumer ( output ) <EOL> inputFile = StringIO ( expectedResult ) <EOL> producer = FileBodyProducer ( <EOL> inputFile , self . cooperator , readSize ) <EOL> complete = producer . startProducing ( consumer ) <EOL> producer . stopProducing ( ) <EOL> self . assertTrue ( inputFile . closed ) <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertEqual ( "<STR_LIT>" , output . getvalue ( ) ) <EOL> self . assertNoResult ( complete ) <EOL> def test_pauseProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedResult = "<STR_LIT>" <EOL> readSize = <NUM_LIT:5> <EOL> output = StringIO ( ) <EOL> consumer = FileConsumer ( output ) <EOL> producer = FileBodyProducer ( <EOL> StringIO ( expectedResult ) , self . cooperator , readSize ) <EOL> complete = producer . startProducing ( consumer ) <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertEqual ( output . getvalue ( ) , expectedResult [ : <NUM_LIT:5> ] ) <EOL> producer . pauseProducing ( ) <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertEqual ( output . getvalue ( ) , expectedResult [ : <NUM_LIT:5> ] ) <EOL> self . assertEqual ( [ ] , self . _scheduled ) <EOL> self . assertNoResult ( complete ) <EOL> def test_resumeProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedResult = "<STR_LIT>" <EOL> readSize = <NUM_LIT:5> <EOL> output = StringIO ( ) <EOL> consumer = FileConsumer ( output ) <EOL> producer = FileBodyProducer ( <EOL> StringIO ( expectedResult ) , self . cooperator , readSize ) <EOL> producer . startProducing ( consumer ) <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertEqual ( expectedResult [ : readSize ] , output . getvalue ( ) ) <EOL> producer . pauseProducing ( ) <EOL> producer . resumeProducing ( ) <EOL> self . _scheduled . pop ( <NUM_LIT:0> ) ( ) <EOL> self . assertEqual ( expectedResult [ : readSize * <NUM_LIT:2> ] , output . getvalue ( ) ) <EOL> class FakeReactorAndConnectMixin : <EOL> """<STR_LIT>""" <EOL> Reactor = MemoryReactorClock <EOL> @ implementer ( IPolicyForHTTPS ) <EOL> class StubPolicy ( object ) : <EOL> """<STR_LIT>""" <EOL> def creatorForNetloc ( self , hostname , port ) : <EOL> """<STR_LIT>""" <EOL> class StubEndpoint ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , endpoint , testCase ) : <EOL> self . endpoint = endpoint <EOL> self . testCase = testCase <EOL> self . factory = _HTTP11ClientFactory ( lambda p : None ) <EOL> self . protocol = StubHTTPProtocol ( ) <EOL> self . factory . buildProtocol = lambda addr : self . protocol <EOL> def connect ( self , ignoredFactory ) : <EOL> self . testCase . protocol = self . protocol <EOL> self . endpoint . connect ( self . factory ) <EOL> return succeed ( self . protocol ) <EOL> def buildAgentForWrapperTest ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent ( reactor , self . StubPolicy ( ) ) <EOL> _oldGetEndpoint = agent . _getEndpoint <EOL> agent . _getEndpoint = lambda * args : ( <EOL> self . StubEndpoint ( _oldGetEndpoint ( * args ) , self ) ) <EOL> return agent <EOL> def connect ( self , factory ) : <EOL> """<STR_LIT>""" <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( None ) <EOL> self . protocol = protocol <EOL> return succeed ( protocol ) <EOL> class DummyEndpoint ( object ) : <EOL> """<STR_LIT>""" <EOL> def connect ( self , factory ) : <EOL> protocol = factory . buildProtocol ( None ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> return succeed ( protocol ) <EOL> class BadEndpoint ( object ) : <EOL> """<STR_LIT>""" <EOL> def connect ( self , factory ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> class DummyFactory ( Factory ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , quiescentCallback ) : <EOL> pass <EOL> protocol = StubHTTPProtocol <EOL> class HTTPConnectionPoolTests ( TestCase , FakeReactorAndConnectMixin ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . fakeReactor = self . Reactor ( ) <EOL> self . pool = HTTPConnectionPool ( self . fakeReactor ) <EOL> self . pool . _factory = DummyFactory <EOL> self . pool . retryAutomatically = False <EOL> def test_getReturnsNewIfCacheEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . pool . _connections , { } ) <EOL> def gotConnection ( conn ) : <EOL> self . assertIsInstance ( conn , StubHTTPProtocol ) <EOL> self . assertNotIn ( conn , self . pool . _connections . values ( ) ) <EOL> unknownKey = <NUM_LIT> <EOL> d = self . pool . getConnection ( unknownKey , DummyEndpoint ( ) ) <EOL> return d . addCallback ( gotConnection ) <EOL> def test_putStartsTimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> self . pool . _putConnection ( ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) , protocol ) <EOL> self . assertEqual ( protocol . transport . disconnecting , False ) <EOL> self . assertIn ( protocol , <EOL> self . pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] ) <EOL> self . fakeReactor . advance ( <NUM_LIT> ) <EOL> self . assertEqual ( protocol . transport . disconnecting , False ) <EOL> self . assertIn ( protocol , <EOL> self . pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] ) <EOL> self . assertIn ( protocol , self . pool . _timeouts ) <EOL> self . fakeReactor . advance ( <NUM_LIT> ) <EOL> self . assertEqual ( protocol . transport . disconnecting , True ) <EOL> self . assertNotIn ( protocol , <EOL> self . pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] ) <EOL> self . assertNotIn ( protocol , self . pool . _timeouts ) <EOL> def test_putExceedsMaxPersistent ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = self . pool <EOL> origCached = [ StubHTTPProtocol ( ) , StubHTTPProtocol ( ) ] <EOL> for p in origCached : <EOL> p . makeConnection ( StringTransport ( ) ) <EOL> pool . _putConnection ( ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) , p ) <EOL> self . assertEqual ( pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] , <EOL> origCached ) <EOL> timeouts = pool . _timeouts . copy ( ) <EOL> newProtocol = StubHTTPProtocol ( ) <EOL> newProtocol . makeConnection ( StringTransport ( ) ) <EOL> pool . _putConnection ( ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) , newProtocol ) <EOL> newCached = pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] <EOL> self . assertEqual ( len ( newCached ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( newCached , [ origCached [ <NUM_LIT:1> ] , newProtocol ] ) <EOL> self . assertEqual ( [ p . transport . disconnecting for p in newCached ] , <EOL> [ False , False ] ) <EOL> self . assertEqual ( origCached [ <NUM_LIT:0> ] . transport . disconnecting , True ) <EOL> self . assertTrue ( timeouts [ origCached [ <NUM_LIT:0> ] ] . cancelled ) <EOL> self . assertNotIn ( origCached [ <NUM_LIT:0> ] , pool . _timeouts ) <EOL> def test_maxPersistentPerHost ( self ) : <EOL> """<STR_LIT>""" <EOL> def addProtocol ( scheme , host , port ) : <EOL> p = StubHTTPProtocol ( ) <EOL> p . makeConnection ( StringTransport ( ) ) <EOL> self . pool . _putConnection ( ( scheme , host , port ) , p ) <EOL> return p <EOL> persistent = [ ] <EOL> persistent . append ( addProtocol ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ) <EOL> persistent . append ( addProtocol ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ) <EOL> addProtocol ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> addProtocol ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> self . pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] , persistent ) <EOL> self . assertEqual ( <EOL> len ( self . pool . _connections [ ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) ] ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( <EOL> len ( self . pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] ) , <NUM_LIT:1> ) <EOL> def test_getCachedConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> self . pool . _putConnection ( ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) , protocol ) <EOL> def gotConnection ( conn ) : <EOL> self . assertIdentical ( protocol , conn ) <EOL> self . assertNotIn ( <EOL> conn , self . pool . _connections [ ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ] ) <EOL> self . fakeReactor . advance ( <NUM_LIT> ) <EOL> self . assertEqual ( conn . transport . disconnecting , False ) <EOL> self . assertNotIn ( conn , self . pool . _timeouts ) <EOL> return self . pool . getConnection ( ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) , <EOL> BadEndpoint ( ) , <EOL> ) . addCallback ( gotConnection ) <EOL> def test_newConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> key = <NUM_LIT> <EOL> self . pool . _putConnection ( key , protocol ) <EOL> def gotConnection ( newConnection ) : <EOL> self . assertNotIdentical ( protocol , newConnection ) <EOL> self . assertIn ( protocol , self . pool . _connections [ key ] ) <EOL> self . assertNotIn ( newConnection , self . pool . _connections . values ( ) ) <EOL> d = self . pool . _newConnection ( key , DummyEndpoint ( ) ) <EOL> return d . addCallback ( gotConnection ) <EOL> def test_getSkipsDisconnected ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = self . pool <EOL> key = ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> origCached = [ StubHTTPProtocol ( ) , StubHTTPProtocol ( ) ] <EOL> for p in origCached : <EOL> p . makeConnection ( StringTransport ( ) ) <EOL> pool . _putConnection ( key , p ) <EOL> self . assertEqual ( pool . _connections [ key ] , origCached ) <EOL> origCached [ <NUM_LIT:0> ] . state = "<STR_LIT>" <EOL> result = [ ] <EOL> self . pool . getConnection ( key , <EOL> BadEndpoint ( ) ) . addCallback ( result . append ) <EOL> self . assertIdentical ( result [ <NUM_LIT:0> ] , origCached [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( pool . _connections [ key ] , [ ] ) <EOL> self . assertEqual ( pool . _timeouts , { } ) <EOL> def test_putNotQuiescent ( self ) : <EOL> """<STR_LIT>""" <EOL> protocol = StubHTTPProtocol ( ) <EOL> self . assertEqual ( protocol . state , "<STR_LIT>" ) <EOL> protocol . state = "<STR_LIT>" <EOL> self . pool . _putConnection ( ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) , protocol ) <EOL> exc , = self . flushLoggedErrors ( RuntimeError ) <EOL> self . assertEqual ( <EOL> exc . value . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertIdentical ( None , self . pool . _connections . get ( <EOL> ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) ) ) <EOL> def test_getUsesQuiescentCallback ( self ) : <EOL> """<STR_LIT>""" <EOL> class StringEndpoint ( object ) : <EOL> def connect ( self , factory ) : <EOL> p = factory . buildProtocol ( None ) <EOL> p . makeConnection ( StringTransport ( ) ) <EOL> return succeed ( p ) <EOL> pool = HTTPConnectionPool ( self . fakeReactor , True ) <EOL> pool . retryAutomatically = False <EOL> result = [ ] <EOL> key = "<STR_LIT>" <EOL> pool . getConnection ( <EOL> key , StringEndpoint ( ) ) . addCallback ( <EOL> result . append ) <EOL> protocol = result [ <NUM_LIT:0> ] <EOL> self . assertIsInstance ( protocol , HTTP11ClientProtocol ) <EOL> protocol . _state = "<STR_LIT>" <EOL> protocol . _quiescentCallback ( protocol ) <EOL> result2 = [ ] <EOL> pool . getConnection ( <EOL> key , StringEndpoint ( ) ) . addCallback ( <EOL> result2 . append ) <EOL> self . assertIdentical ( result2 [ <NUM_LIT:0> ] , protocol ) <EOL> def test_closeCachedConnections ( self ) : <EOL> """<STR_LIT>""" <EOL> persistent = [ ] <EOL> def addProtocol ( scheme , host , port ) : <EOL> p = HTTP11ClientProtocol ( ) <EOL> p . makeConnection ( StringTransport ( ) ) <EOL> self . pool . _putConnection ( ( scheme , host , port ) , p ) <EOL> persistent . append ( p ) <EOL> addProtocol ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> addProtocol ( "<STR_LIT:http>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> doneDeferred = self . pool . closeCachedConnections ( ) <EOL> for p in persistent : <EOL> self . assertEqual ( p . transport . disconnecting , True ) <EOL> self . assertEqual ( self . pool . _connections , { } ) <EOL> for dc in self . fakeReactor . getDelayedCalls ( ) : <EOL> self . assertEqual ( dc . cancelled , True ) <EOL> self . assertEqual ( self . pool . _timeouts , { } ) <EOL> result = [ ] <EOL> doneDeferred . addCallback ( result . append ) <EOL> self . assertEqual ( result , [ ] ) <EOL> persistent [ <NUM_LIT:0> ] . connectionLost ( Failure ( ConnectionDone ( ) ) ) <EOL> self . assertEqual ( result , [ ] ) <EOL> persistent [ <NUM_LIT:1> ] . connectionLost ( Failure ( ConnectionDone ( ) ) ) <EOL> self . assertEqual ( result , [ None ] ) <EOL> def test_cancelGetConnectionCancelsEndpointConnect ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . pool . _connections , { } ) <EOL> connectionResult = Deferred ( ) <EOL> class Endpoint : <EOL> def connect ( self , factory ) : <EOL> return connectionResult <EOL> d = self . pool . getConnection ( <NUM_LIT> , Endpoint ( ) ) <EOL> d . cancel ( ) <EOL> self . assertEqual ( self . failureResultOf ( connectionResult ) . type , <EOL> CancelledError ) <EOL> class AgentTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( verifyObject ( IAgent , self . makeAgent ( ) ) ) <EOL> @ implementer ( IAgentEndpointFactory ) <EOL> class StubEndpointFactory ( object ) : <EOL> """<STR_LIT>""" <EOL> def endpointForURI ( self , uri ) : <EOL> """<STR_LIT>""" <EOL> return ( uri . scheme , uri . host , uri . port ) <EOL> class AgentTests ( TestCase , FakeReactorAndConnectMixin , AgentTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeAgent ( self ) : <EOL> """<STR_LIT>""" <EOL> return client . Agent ( self . reactor ) <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . reactor = self . Reactor ( ) <EOL> self . agent = self . makeAgent ( ) <EOL> def test_defaultPool ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent ( self . reactor ) <EOL> self . assertIsInstance ( agent . _pool , HTTPConnectionPool ) <EOL> self . assertEqual ( agent . _pool . persistent , False ) <EOL> self . assertIdentical ( agent . _reactor , agent . _pool . _reactor ) <EOL> def test_persistent ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = HTTPConnectionPool ( self . reactor ) <EOL> agent = client . Agent ( self . reactor , pool = pool ) <EOL> agent . _getEndpoint = lambda * args : self <EOL> agent . request ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( self . protocol . requests [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . persistent , True ) <EOL> def test_nonPersistent ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = HTTPConnectionPool ( self . reactor , persistent = False ) <EOL> agent = client . Agent ( self . reactor , pool = pool ) <EOL> agent . _getEndpoint = lambda * args : self <EOL> agent . request ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( self . protocol . requests [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . persistent , False ) <EOL> def test_connectUsesConnectionPool ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = DummyEndpoint ( ) <EOL> class MyAgent ( client . Agent ) : <EOL> def _getEndpoint ( this , uri ) : <EOL> self . assertEqual ( ( uri . scheme , uri . host , uri . port ) , <EOL> ( "<STR_LIT:http>" , "<STR_LIT:foo>" , <NUM_LIT> ) ) <EOL> return endpoint <EOL> class DummyPool ( object ) : <EOL> connected = False <EOL> persistent = False <EOL> def getConnection ( this , key , ep ) : <EOL> this . connected = True <EOL> self . assertEqual ( ep , endpoint ) <EOL> self . assertEqual ( key , ( "<STR_LIT:http>" , "<STR_LIT:foo>" , <NUM_LIT> ) ) <EOL> return defer . succeed ( StubHTTPProtocol ( ) ) <EOL> pool = DummyPool ( ) <EOL> agent = MyAgent ( self . reactor , pool = pool ) <EOL> self . assertIdentical ( pool , agent . _pool ) <EOL> headers = http_headers . Headers ( ) <EOL> headers . addRawHeader ( "<STR_LIT:host>" , "<STR_LIT:foo>" ) <EOL> bodyProducer = object ( ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' , <EOL> bodyProducer = bodyProducer , headers = headers ) <EOL> self . assertEqual ( agent . _pool . connected , True ) <EOL> def test_unsupportedScheme ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . assertFailure ( <EOL> self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) , <EOL> SchemeNotSupported ) <EOL> def test_connectionFailed ( self ) : <EOL> """<STR_LIT>""" <EOL> result = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> host , port , factory = self . reactor . tcpClients . pop ( ) [ : <NUM_LIT:3> ] <EOL> factory . clientConnectionFailed ( None , Failure ( ConnectionRefusedError ( ) ) ) <EOL> return self . assertFailure ( result , ConnectionRefusedError ) <EOL> def test_connectHTTP ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedHost = '<STR_LIT>' <EOL> expectedPort = <NUM_LIT> <EOL> endpoint = self . agent . _getEndpoint ( <EOL> URI . fromBytes ( b'<STR_LIT>' % ( expectedHost , expectedPort ) ) ) <EOL> self . assertEqual ( endpoint . _host , expectedHost ) <EOL> self . assertEqual ( endpoint . _port , expectedPort ) <EOL> self . assertIsInstance ( endpoint , TCP4ClientEndpoint ) <EOL> def test_connectHTTPSCustomContextFactory ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedHost = '<STR_LIT>' <EOL> expectedPort = <NUM_LIT> <EOL> expectedContext = object ( ) <EOL> contextArgs = [ ] <EOL> class StubWebContextFactory ( object ) : <EOL> def getContext ( self , hostname , port ) : <EOL> contextArgs . append ( ( hostname , port ) ) <EOL> return expectedContext <EOL> agent = client . Agent ( self . reactor , StubWebContextFactory ( ) ) <EOL> endpoint = agent . _getEndpoint ( <EOL> URI . fromBytes ( b'<STR_LIT>' % ( expectedHost , expectedPort ) ) ) <EOL> contextFactory = endpoint . _sslContextFactory <EOL> context = contextFactory . getContext ( ) <EOL> self . assertEqual ( context , expectedContext ) <EOL> self . assertEqual ( contextArgs , [ ( expectedHost , expectedPort ) ] ) <EOL> def test_hostProvided ( self ) : <EOL> """<STR_LIT>""" <EOL> self . agent . _getEndpoint = lambda * args : self <EOL> self . agent . request ( <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( req . headers . getRawHeaders ( '<STR_LIT:host>' ) , [ '<STR_LIT>' ] ) <EOL> def test_hostOverride ( self ) : <EOL> """<STR_LIT>""" <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , '<STR_LIT:host>' : [ '<STR_LIT>' ] } ) <EOL> self . agent . _getEndpoint = lambda * args : self <EOL> self . agent . request ( <EOL> '<STR_LIT:GET>' , '<STR_LIT>' , headers ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( req . headers . getRawHeaders ( '<STR_LIT:host>' ) , [ '<STR_LIT>' ] ) <EOL> def test_headersUnmodified ( self ) : <EOL> """<STR_LIT>""" <EOL> headers = http_headers . Headers ( ) <EOL> self . agent . _getEndpoint = lambda * args : self <EOL> self . agent . request ( <EOL> '<STR_LIT:GET>' , '<STR_LIT>' , headers ) <EOL> protocol = self . protocol <EOL> self . assertEqual ( len ( protocol . requests ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( headers , http_headers . Headers ( ) ) <EOL> def test_hostValueStandardHTTP ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . agent . _computeHostValue ( '<STR_LIT:http>' , '<STR_LIT>' , <NUM_LIT> ) , <EOL> '<STR_LIT>' ) <EOL> def test_hostValueNonStandardHTTP ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . agent . _computeHostValue ( '<STR_LIT:http>' , '<STR_LIT>' , <NUM_LIT> ) , <EOL> '<STR_LIT>' ) <EOL> def test_hostValueStandardHTTPS ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . agent . _computeHostValue ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) , <EOL> '<STR_LIT>' ) <EOL> def test_hostValueNonStandardHTTPS ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . agent . _computeHostValue ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) , <EOL> '<STR_LIT>' ) <EOL> def test_request ( self ) : <EOL> """<STR_LIT>""" <EOL> self . agent . _getEndpoint = lambda * args : self <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] } ) <EOL> body = object ( ) <EOL> self . agent . request ( <EOL> '<STR_LIT:GET>' , '<STR_LIT>' , headers , body ) <EOL> protocol = self . protocol <EOL> self . assertEqual ( len ( protocol . requests ) , <NUM_LIT:1> ) <EOL> req , res = protocol . requests . pop ( ) <EOL> self . assertIsInstance ( req , Request ) <EOL> self . assertEqual ( req . method , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( req . uri , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> req . headers , <EOL> http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT:host>' : [ '<STR_LIT>' ] } ) ) <EOL> self . assertIdentical ( req . bodyProducer , body ) <EOL> def test_connectTimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent ( self . reactor , connectTimeout = <NUM_LIT:5> ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> timeout = self . reactor . tcpClients . pop ( ) [ <NUM_LIT:3> ] <EOL> self . assertEqual ( <NUM_LIT:5> , timeout ) <EOL> def test_connectSSLTimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent ( self . reactor , self . StubPolicy ( ) , connectTimeout = <NUM_LIT:5> ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> timeout = self . reactor . sslClients . pop ( ) [ <NUM_LIT:4> ] <EOL> self . assertEqual ( <NUM_LIT:5> , timeout ) <EOL> def test_bindAddress ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent ( self . reactor , bindAddress = '<STR_LIT>' ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> address = self . reactor . tcpClients . pop ( ) [ <NUM_LIT:4> ] <EOL> self . assertEqual ( '<STR_LIT>' , address ) <EOL> def test_bindAddressSSL ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent ( self . reactor , self . StubPolicy ( ) , <EOL> bindAddress = '<STR_LIT>' ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> address = self . reactor . sslClients . pop ( ) [ <NUM_LIT:5> ] <EOL> self . assertEqual ( '<STR_LIT>' , address ) <EOL> def test_responseIncludesRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = b'<STR_LIT>' <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> d = agent . request ( '<STR_LIT:GET>' , uri ) <EOL> self . assertEqual ( len ( self . protocol . requests ) , <NUM_LIT:1> ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertIsInstance ( req , Request ) <EOL> resp = client . Response . _construct ( <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <EOL> <NUM_LIT:200> , <EOL> '<STR_LIT:OK>' , <EOL> client . Headers ( { } ) , <EOL> None , <EOL> req ) <EOL> res . callback ( resp ) <EOL> response = self . successResultOf ( d ) <EOL> self . assertEqual ( <EOL> ( response . request . method , response . request . absoluteURI , <EOL> response . request . headers ) , <EOL> ( req . method , req . absoluteURI , req . headers ) ) <EOL> def test_requestAbsoluteURI ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = b'<STR_LIT>' <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> agent . request ( b'<STR_LIT:GET>' , uri ) <EOL> self . assertEqual ( len ( self . protocol . requests ) , <NUM_LIT:1> ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertIsInstance ( req , Request ) <EOL> self . assertEquals ( req . absoluteURI , uri ) <EOL> def test_requestMissingAbsoluteURI ( self ) : <EOL> """<STR_LIT>""" <EOL> request = client . Request ( b'<STR_LIT>' , b'<STR_LIT:/>' , client . Headers ( ) , None ) <EOL> self . assertIdentical ( request . absoluteURI , None ) <EOL> def test_endpointFactory ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = StubEndpointFactory ( ) <EOL> agent = client . Agent . usingEndpointFactory ( <EOL> None , endpointFactory = factory ) <EOL> uri = URI . fromBytes ( b'<STR_LIT>' ) <EOL> returnedEndpoint = agent . _getEndpoint ( uri ) <EOL> self . assertEqual ( returnedEndpoint , ( b"<STR_LIT:http>" , b"<STR_LIT>" , <NUM_LIT> ) ) <EOL> def test_endpointFactoryDefaultPool ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . Agent . usingEndpointFactory ( <EOL> self . reactor , StubEndpointFactory ( ) ) <EOL> pool = agent . _pool <EOL> self . assertEqual ( ( pool . __class__ , pool . persistent , pool . _reactor ) , <EOL> ( HTTPConnectionPool , False , agent . _reactor ) ) <EOL> def test_endpointFactoryPool ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = object ( ) <EOL> agent = client . Agent . usingEndpointFactory ( <EOL> self . reactor , StubEndpointFactory ( ) , pool ) <EOL> self . assertIs ( pool , agent . _pool ) <EOL> class AgentHTTPSTests ( TestCase , FakeReactorAndConnectMixin ) : <EOL> """<STR_LIT>""" <EOL> if ssl is None : <EOL> skip = "<STR_LIT>" <EOL> def makeEndpoint ( self , host = '<STR_LIT>' , port = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> return client . Agent ( self . Reactor ( ) ) . _getEndpoint ( <EOL> URI . fromBytes ( b'<STR_LIT>' % ( host , port ) ) ) <EOL> def test_endpointType ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertIsInstance ( self . makeEndpoint ( ) , SSL4ClientEndpoint ) <EOL> def test_hostArgumentIsRespected ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedHost = '<STR_LIT>' <EOL> endpoint = self . makeEndpoint ( host = expectedHost ) <EOL> self . assertEqual ( endpoint . _host , expectedHost ) <EOL> def test_portArgumentIsRespected ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedPort = <NUM_LIT> <EOL> endpoint = self . makeEndpoint ( port = expectedPort ) <EOL> self . assertEqual ( endpoint . _port , expectedPort ) <EOL> def test_contextFactoryType ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = self . makeEndpoint ( ) <EOL> contextFactory = endpoint . _sslContextFactory <EOL> self . assertIsInstance ( contextFactory , ClientTLSOptions ) <EOL> self . assertEqual ( contextFactory . _hostname , u"<STR_LIT>" ) <EOL> def test_connectHTTPSCustomConnectionCreator ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedHost = b'<STR_LIT>' <EOL> expectedPort = <NUM_LIT> <EOL> class JustEnoughConnection ( object ) : <EOL> handshakeStarted = False <EOL> connectState = False <EOL> def do_handshake ( self ) : <EOL> """<STR_LIT>""" <EOL> self . handshakeStarted = True <EOL> def set_connect_state ( self ) : <EOL> """<STR_LIT>""" <EOL> self . connectState = True <EOL> contextArgs = [ ] <EOL> @ implementer ( IOpenSSLClientConnectionCreator ) <EOL> class JustEnoughCreator ( object ) : <EOL> def __init__ ( self , hostname , port ) : <EOL> self . hostname = hostname <EOL> self . port = port <EOL> def clientConnectionForTLS ( self , tlsProtocol ) : <EOL> """<STR_LIT>""" <EOL> contextArgs . append ( ( tlsProtocol , self . hostname , self . port ) ) <EOL> return expectedConnection <EOL> expectedConnection = JustEnoughConnection ( ) <EOL> @ implementer ( IPolicyForHTTPS ) <EOL> class StubBrowserLikePolicyForHTTPS ( object ) : <EOL> def creatorForNetloc ( self , hostname , port ) : <EOL> """<STR_LIT>""" <EOL> return JustEnoughCreator ( hostname , port ) <EOL> expectedCreatorCreator = StubBrowserLikePolicyForHTTPS ( ) <EOL> reactor = self . Reactor ( ) <EOL> agent = client . Agent ( reactor , expectedCreatorCreator ) <EOL> endpoint = agent . _getEndpoint ( <EOL> URI . fromBytes ( b'<STR_LIT>' % ( expectedHost , expectedPort ) ) ) <EOL> endpoint . connect ( Factory . forProtocol ( Protocol ) ) <EOL> passedFactory = reactor . sslClients [ - <NUM_LIT:1> ] [ <NUM_LIT:2> ] <EOL> passedContextFactory = reactor . sslClients [ - <NUM_LIT:1> ] [ <NUM_LIT:3> ] <EOL> tlsFactory = TLSMemoryBIOFactory ( <EOL> passedContextFactory , True , passedFactory <EOL> ) <EOL> tlsProtocol = tlsFactory . buildProtocol ( None ) <EOL> tlsProtocol . makeConnection ( StringTransport ( ) ) <EOL> tls = contextArgs [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertIsInstance ( tls , TLSMemoryBIOProtocol ) <EOL> self . assertEqual ( contextArgs [ <NUM_LIT:0> ] [ <NUM_LIT:1> : ] , ( expectedHost , expectedPort ) ) <EOL> self . assertTrue ( expectedConnection . handshakeStarted ) <EOL> self . assertTrue ( expectedConnection . connectState ) <EOL> def test_deprecatedDuckPolicy ( self ) : <EOL> """<STR_LIT>""" <EOL> def warnMe ( ) : <EOL> client . Agent ( MemoryReactorClock ( ) , <EOL> "<STR_LIT>" ) <EOL> warnMe ( ) <EOL> warnings = self . flushWarnings ( [ warnMe ] ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> [ warning ] = warnings <EOL> self . assertEqual ( warning [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warning [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_alternateTrustRoot ( self ) : <EOL> """<STR_LIT>""" <EOL> @ implementer ( IOpenSSLTrustRoot ) <EOL> class CustomOpenSSLTrustRoot ( object ) : <EOL> called = False <EOL> context = None <EOL> def _addCACertsToContext ( self , context ) : <EOL> self . called = True <EOL> self . context = context <EOL> trustRoot = CustomOpenSSLTrustRoot ( ) <EOL> policy = BrowserLikePolicyForHTTPS ( trustRoot = trustRoot ) <EOL> creator = policy . creatorForNetloc ( b"<STR_LIT>" , <NUM_LIT> ) <EOL> self . assertTrue ( trustRoot . called ) <EOL> connection = creator . clientConnectionForTLS ( None ) <EOL> self . assertIs ( trustRoot . context , connection . get_context ( ) ) <EOL> class WebClientContextFactoryTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> from twisted . web . client import WebClientContextFactory <EOL> self . warned = self . flushWarnings ( [ WebClientContextFactoryTests . setUp ] ) <EOL> self . webClientContextFactory = WebClientContextFactory <EOL> def test_deprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( len ( self . warned ) , <NUM_LIT:1> ) <EOL> [ warning ] = self . warned <EOL> self . assertEqual ( warning [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warning [ '<STR_LIT:message>' ] , <EOL> getDeprecationWarningString ( <EOL> self . webClientContextFactory , Version ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> replacement = BrowserLikePolicyForHTTPS , <EOL> ) <EOL> . replace ( "<STR_LIT:;>" , "<STR_LIT::>" ) <EOL> ) <EOL> def test_missingSSL ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> self . webClientContextFactory ( ) . getContext , <EOL> '<STR_LIT>' , <NUM_LIT> , <EOL> ) <EOL> def test_returnsContext ( self ) : <EOL> """<STR_LIT>""" <EOL> ctx = self . webClientContextFactory ( ) . getContext ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . assertIsInstance ( ctx , ssl . SSL . Context ) <EOL> def test_setsTrustRootOnContextToDefaultTrustRoot ( self ) : <EOL> """<STR_LIT>""" <EOL> ctx = self . webClientContextFactory ( ) <EOL> certificateOptions = ctx . _getCertificateOptions ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . assertIsInstance ( <EOL> certificateOptions . trustRoot , ssl . OpenSSLDefaultPaths ) <EOL> if ssl is None : <EOL> test_returnsContext . skip = "<STR_LIT>" <EOL> test_setsTrustRootOnContextToDefaultTrustRoot . skip = ( <EOL> "<STR_LIT>" ) <EOL> else : <EOL> test_missingSSL . skip = "<STR_LIT>" <EOL> class HTTPConnectionPoolRetryTests ( TestCase , FakeReactorAndConnectMixin ) : <EOL> """<STR_LIT>""" <EOL> def test_onlyRetryIdempotentMethods ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( None ) <EOL> connection = client . _RetryingHTTP11ClientProtocol ( None , pool ) <EOL> self . assertTrue ( connection . _shouldRetry ( "<STR_LIT:GET>" , RequestNotSent ( ) , None ) ) <EOL> self . assertTrue ( connection . _shouldRetry ( "<STR_LIT>" , RequestNotSent ( ) , None ) ) <EOL> self . assertTrue ( connection . _shouldRetry ( <EOL> "<STR_LIT>" , RequestNotSent ( ) , None ) ) <EOL> self . assertTrue ( connection . _shouldRetry ( <EOL> "<STR_LIT>" , RequestNotSent ( ) , None ) ) <EOL> self . assertTrue ( connection . _shouldRetry ( <EOL> "<STR_LIT>" , RequestNotSent ( ) , None ) ) <EOL> self . assertFalse ( connection . _shouldRetry ( <EOL> "<STR_LIT:POST>" , RequestNotSent ( ) , None ) ) <EOL> self . assertFalse ( connection . _shouldRetry ( <EOL> "<STR_LIT>" , RequestNotSent ( ) , None ) ) <EOL> def test_onlyRetryIfNoResponseReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( None ) <EOL> connection = client . _RetryingHTTP11ClientProtocol ( None , pool ) <EOL> self . assertTrue ( connection . _shouldRetry ( "<STR_LIT:GET>" , RequestNotSent ( ) , None ) ) <EOL> self . assertTrue ( connection . _shouldRetry ( <EOL> "<STR_LIT:GET>" , RequestTransmissionFailed ( [ ] ) , None ) ) <EOL> self . assertTrue ( connection . _shouldRetry ( <EOL> "<STR_LIT:GET>" , ResponseNeverReceived ( [ ] ) , None ) ) <EOL> self . assertFalse ( connection . _shouldRetry ( <EOL> "<STR_LIT:GET>" , ResponseFailed ( [ ] ) , None ) ) <EOL> self . assertFalse ( connection . _shouldRetry ( <EOL> "<STR_LIT:GET>" , ConnectionRefusedError ( ) , None ) ) <EOL> def test_dontRetryIfFailedDueToCancel ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( None ) <EOL> connection = client . _RetryingHTTP11ClientProtocol ( None , pool ) <EOL> exception = ResponseNeverReceived ( [ Failure ( defer . CancelledError ( ) ) ] ) <EOL> self . assertFalse ( connection . _shouldRetry ( <EOL> "<STR_LIT:GET>" , exception , None ) ) <EOL> def test_retryIfFailedDueToNonCancelException ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( None ) <EOL> connection = client . _RetryingHTTP11ClientProtocol ( None , pool ) <EOL> self . assertTrue ( connection . _shouldRetry ( <EOL> "<STR_LIT:GET>" , ResponseNeverReceived ( [ Failure ( Exception ( ) ) ] ) , None ) ) <EOL> def test_wrappedOnPersistentReturned ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( Clock ( ) ) <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> pool . _putConnection ( <NUM_LIT> , protocol ) <EOL> d = pool . getConnection ( <NUM_LIT> , DummyEndpoint ( ) ) <EOL> def gotConnection ( connection ) : <EOL> self . assertIsInstance ( connection , <EOL> client . _RetryingHTTP11ClientProtocol ) <EOL> self . assertIdentical ( connection . _clientProtocol , protocol ) <EOL> return d . addCallback ( gotConnection ) <EOL> def test_notWrappedOnNewReturned ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( None ) <EOL> d = pool . getConnection ( <NUM_LIT> , DummyEndpoint ( ) ) <EOL> def gotConnection ( connection ) : <EOL> self . assertIdentical ( connection . __class__ , HTTP11ClientProtocol ) <EOL> return d . addCallback ( gotConnection ) <EOL> def retryAttempt ( self , willWeRetry ) : <EOL> """<STR_LIT>""" <EOL> protocols = [ ] <EOL> def newProtocol ( ) : <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocols . append ( protocol ) <EOL> return defer . succeed ( protocol ) <EOL> bodyProducer = object ( ) <EOL> request = client . Request ( "<STR_LIT>" , "<STR_LIT:/>" , client . Headers ( ) , bodyProducer , <EOL> persistent = True ) <EOL> newProtocol ( ) <EOL> protocol = protocols [ <NUM_LIT:0> ] <EOL> retrier = client . _RetryingHTTP11ClientProtocol ( protocol , newProtocol ) <EOL> def _shouldRetry ( m , e , bp ) : <EOL> self . assertEqual ( m , "<STR_LIT>" ) <EOL> self . assertIdentical ( bp , bodyProducer ) <EOL> self . assertIsInstance ( e , ( RequestNotSent , ResponseNeverReceived ) ) <EOL> return willWeRetry <EOL> retrier . _shouldRetry = _shouldRetry <EOL> d = retrier . request ( request ) <EOL> self . assertEqual ( len ( protocols ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( protocols [ <NUM_LIT:0> ] . requests ) , <NUM_LIT:1> ) <EOL> protocol . requests [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . errback ( RequestNotSent ( ) ) <EOL> return d , protocols <EOL> def test_retryIfShouldRetryReturnsTrue ( self ) : <EOL> """<STR_LIT>""" <EOL> d , protocols = self . retryAttempt ( True ) <EOL> self . assertEqual ( len ( protocols ) , <NUM_LIT:2> ) <EOL> response = object ( ) <EOL> protocols [ <NUM_LIT:1> ] . requests [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . callback ( response ) <EOL> return d . addCallback ( self . assertIdentical , response ) <EOL> def test_dontRetryIfShouldRetryReturnsFalse ( self ) : <EOL> """<STR_LIT>""" <EOL> d , protocols = self . retryAttempt ( False ) <EOL> self . assertEqual ( len ( protocols ) , <NUM_LIT:1> ) <EOL> return self . assertFailure ( d , RequestNotSent ) <EOL> def test_onlyRetryWithoutBody ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( None ) <EOL> connection = client . _RetryingHTTP11ClientProtocol ( None , pool ) <EOL> self . assertTrue ( connection . _shouldRetry ( "<STR_LIT:GET>" , RequestNotSent ( ) , None ) ) <EOL> self . assertFalse ( connection . _shouldRetry ( "<STR_LIT:GET>" , RequestNotSent ( ) , object ( ) ) ) <EOL> def test_onlyRetryOnce ( self ) : <EOL> """<STR_LIT>""" <EOL> d , protocols = self . retryAttempt ( True ) <EOL> self . assertEqual ( len ( protocols ) , <NUM_LIT:2> ) <EOL> protocols [ <NUM_LIT:1> ] . requests [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . errback ( ResponseNeverReceived ( [ ] ) ) <EOL> self . assertEqual ( len ( protocols ) , <NUM_LIT:2> ) <EOL> return self . assertFailure ( d , ResponseNeverReceived ) <EOL> def test_dontRetryIfRetryAutomaticallyFalse ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( Clock ( ) ) <EOL> pool . retryAutomatically = False <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> pool . _putConnection ( <NUM_LIT> , protocol ) <EOL> d = pool . getConnection ( <NUM_LIT> , DummyEndpoint ( ) ) <EOL> def gotConnection ( connection ) : <EOL> self . assertIdentical ( connection , protocol ) <EOL> return d . addCallback ( gotConnection ) <EOL> def test_retryWithNewConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = client . HTTPConnectionPool ( Clock ( ) ) <EOL> key = <NUM_LIT> <EOL> endpoint = DummyEndpoint ( ) <EOL> newConnections = [ ] <EOL> def newConnection ( k , e ) : <EOL> newConnections . append ( ( k , e ) ) <EOL> pool . _newConnection = newConnection <EOL> protocol = StubHTTPProtocol ( ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> pool . _putConnection ( key , protocol ) <EOL> d = pool . getConnection ( key , endpoint ) <EOL> def gotConnection ( connection ) : <EOL> self . assertIsInstance ( connection , <EOL> client . _RetryingHTTP11ClientProtocol ) <EOL> self . assertIdentical ( connection . _clientProtocol , protocol ) <EOL> self . assertEqual ( newConnections , [ ] ) <EOL> connection . _newConnection ( ) <EOL> self . assertEqual ( len ( newConnections ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( newConnections [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , key ) <EOL> self . assertIdentical ( newConnections [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , endpoint ) <EOL> return d . addCallback ( gotConnection ) <EOL> class CookieTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def addCookies ( self , cookieJar , uri , cookies ) : <EOL> """<STR_LIT>""" <EOL> response = client . _FakeUrllib2Response ( <EOL> client . Response ( <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <EOL> <NUM_LIT:200> , <EOL> '<STR_LIT:OK>' , <EOL> client . Headers ( { '<STR_LIT>' : cookies } ) , <EOL> None ) ) <EOL> request = client . _FakeUrllib2Request ( uri ) <EOL> cookieJar . extract_cookies ( response , request ) <EOL> return request , response <EOL> class CookieJarTests ( TestCase , CookieTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeCookieJar ( self ) : <EOL> """<STR_LIT>""" <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> reqres = self . addCookies ( <EOL> cookieJar , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> return cookieJar , reqres <EOL> def test_extractCookies ( self ) : <EOL> """<STR_LIT>""" <EOL> jar = self . makeCookieJar ( ) [ <NUM_LIT:0> ] <EOL> cookies = dict ( [ ( c . name , c ) for c in jar ] ) <EOL> cookie = cookies [ '<STR_LIT:foo>' ] <EOL> self . assertEqual ( cookie . version , <NUM_LIT:0> ) <EOL> self . assertEqual ( cookie . name , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( cookie . value , '<STR_LIT:1>' ) <EOL> self . assertEqual ( cookie . path , '<STR_LIT>' ) <EOL> self . assertEqual ( cookie . comment , '<STR_LIT:hello>' ) <EOL> self . assertEqual ( cookie . get_nonstandard_attr ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> cookie = cookies [ '<STR_LIT:bar>' ] <EOL> self . assertEqual ( cookie . version , <NUM_LIT:0> ) <EOL> self . assertEqual ( cookie . name , '<STR_LIT:bar>' ) <EOL> self . assertEqual ( cookie . value , '<STR_LIT:2>' ) <EOL> self . assertEqual ( cookie . path , '<STR_LIT:/>' ) <EOL> self . assertEqual ( cookie . comment , '<STR_LIT>' ) <EOL> self . assertIdentical ( cookie . get_nonstandard_attr ( '<STR_LIT>' ) , None ) <EOL> def test_sendCookie ( self ) : <EOL> """<STR_LIT>""" <EOL> jar , ( request , response ) = self . makeCookieJar ( ) <EOL> self . assertIdentical ( <EOL> request . get_header ( '<STR_LIT>' , None ) , <EOL> None ) <EOL> jar . add_cookie_header ( request ) <EOL> self . assertEqual ( <EOL> request . get_header ( '<STR_LIT>' , None ) , <EOL> '<STR_LIT>' ) <EOL> class CookieAgentTests ( TestCase , CookieTestsMixin , FakeReactorAndConnectMixin , <EOL> AgentTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeAgent ( self ) : <EOL> """<STR_LIT>""" <EOL> return client . CookieAgent ( <EOL> self . buildAgentForWrapperTest ( self . reactor ) , <EOL> cookielib . CookieJar ( ) ) <EOL> def setUp ( self ) : <EOL> self . reactor = self . Reactor ( ) <EOL> def test_emptyCookieJarRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> self . assertEqual ( list ( cookieJar ) , [ ] ) <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> cookieAgent = client . CookieAgent ( agent , cookieJar ) <EOL> d = cookieAgent . request ( <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> def _checkCookie ( ignored ) : <EOL> cookies = list ( cookieJar ) <EOL> self . assertEqual ( len ( cookies ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( cookies [ <NUM_LIT:0> ] . name , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( cookies [ <NUM_LIT:0> ] . value , '<STR_LIT:1>' ) <EOL> d . addCallback ( _checkCookie ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertIdentical ( req . headers . getRawHeaders ( '<STR_LIT>' ) , None ) <EOL> resp = client . Response ( <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <EOL> <NUM_LIT:200> , <EOL> '<STR_LIT:OK>' , <EOL> client . Headers ( { '<STR_LIT>' : [ '<STR_LIT>' , ] } ) , <EOL> None ) <EOL> res . callback ( resp ) <EOL> return d <EOL> def test_requestWithCookie ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' <EOL> cookie = '<STR_LIT>' <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> self . addCookies ( cookieJar , uri , [ cookie ] ) <EOL> self . assertEqual ( len ( list ( cookieJar ) ) , <NUM_LIT:1> ) <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> cookieAgent = client . CookieAgent ( agent , cookieJar ) <EOL> cookieAgent . request ( '<STR_LIT:GET>' , uri ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( req . headers . getRawHeaders ( '<STR_LIT>' ) , [ cookie ] ) <EOL> def test_secureCookie ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' <EOL> cookie = '<STR_LIT>' <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> self . addCookies ( cookieJar , uri , [ cookie ] ) <EOL> self . assertEqual ( len ( list ( cookieJar ) ) , <NUM_LIT:1> ) <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> cookieAgent = client . CookieAgent ( agent , cookieJar ) <EOL> cookieAgent . request ( '<STR_LIT:GET>' , uri ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( req . headers . getRawHeaders ( '<STR_LIT>' ) , [ '<STR_LIT>' ] ) <EOL> def test_secureCookieOnInsecureConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' <EOL> cookie = '<STR_LIT>' <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> self . addCookies ( cookieJar , uri , [ cookie ] ) <EOL> self . assertEqual ( len ( list ( cookieJar ) ) , <NUM_LIT:1> ) <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> cookieAgent = client . CookieAgent ( agent , cookieJar ) <EOL> cookieAgent . request ( '<STR_LIT:GET>' , uri ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertIdentical ( None , req . headers . getRawHeaders ( '<STR_LIT>' ) ) <EOL> def test_portCookie ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' <EOL> cookie = '<STR_LIT>' <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> self . addCookies ( cookieJar , uri , [ cookie ] ) <EOL> self . assertEqual ( len ( list ( cookieJar ) ) , <NUM_LIT:1> ) <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> cookieAgent = client . CookieAgent ( agent , cookieJar ) <EOL> cookieAgent . request ( '<STR_LIT:GET>' , uri ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( req . headers . getRawHeaders ( '<STR_LIT>' ) , [ '<STR_LIT>' ] ) <EOL> def test_portCookieOnWrongPort ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' <EOL> cookie = '<STR_LIT>' <EOL> cookieJar = cookielib . CookieJar ( ) <EOL> self . addCookies ( cookieJar , uri , [ cookie ] ) <EOL> self . assertEqual ( len ( list ( cookieJar ) ) , <NUM_LIT:0> ) <EOL> class Decoder1 ( proxyForInterface ( IResponse ) ) : <EOL> """<STR_LIT>""" <EOL> class Decoder2 ( Decoder1 ) : <EOL> """<STR_LIT>""" <EOL> class ContentDecoderAgentTests ( TestCase , FakeReactorAndConnectMixin , <EOL> AgentTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeAgent ( self ) : <EOL> """<STR_LIT>""" <EOL> return client . ContentDecoderAgent ( self . agent , [ ] ) <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . reactor = self . Reactor ( ) <EOL> self . agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> def test_acceptHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . ContentDecoderAgent ( <EOL> self . agent , [ ( '<STR_LIT>' , Decoder1 ) , ( '<STR_LIT>' , Decoder2 ) ] ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> protocol = self . protocol <EOL> self . assertEqual ( len ( protocol . requests ) , <NUM_LIT:1> ) <EOL> req , res = protocol . requests . pop ( ) <EOL> self . assertEqual ( req . headers . getRawHeaders ( '<STR_LIT>' ) , <EOL> [ '<STR_LIT>' ] ) <EOL> def test_existingHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> agent = client . ContentDecoderAgent ( <EOL> self . agent , [ ( '<STR_LIT>' , Decoder1 ) , ( '<STR_LIT>' , Decoder2 ) ] ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' , headers = headers ) <EOL> protocol = self . protocol <EOL> self . assertEqual ( len ( protocol . requests ) , <NUM_LIT:1> ) <EOL> req , res = protocol . requests . pop ( ) <EOL> self . assertEqual ( <EOL> list ( sorted ( req . headers . getAllRawHeaders ( ) ) ) , <EOL> [ ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( '<STR_LIT>' , [ '<STR_LIT:bar>' ] ) , <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ] ) <EOL> def test_plainEncodingResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . ContentDecoderAgent ( <EOL> self . agent , [ ( '<STR_LIT>' , Decoder1 ) , ( '<STR_LIT>' , Decoder2 ) ] ) <EOL> deferred = agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , http_headers . Headers ( ) , <EOL> None ) <EOL> res . callback ( response ) <EOL> return deferred . addCallback ( self . assertIdentical , response ) <EOL> def test_unsupportedEncoding ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . ContentDecoderAgent ( <EOL> self . agent , [ ( '<STR_LIT>' , Decoder1 ) , ( '<STR_LIT>' , Decoder2 ) ] ) <EOL> deferred = agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> return deferred . addCallback ( self . assertIdentical , response ) <EOL> def test_unknownEncoding ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = client . ContentDecoderAgent ( <EOL> self . agent , [ ( '<STR_LIT>' , Decoder1 ) , ( '<STR_LIT>' , Decoder2 ) ] ) <EOL> deferred = agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT>' : <EOL> [ '<STR_LIT>' ] } ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> def check ( result ) : <EOL> self . assertNotIdentical ( response , result ) <EOL> self . assertIsInstance ( result , Decoder2 ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] , <EOL> result . headers . getRawHeaders ( '<STR_LIT>' ) ) <EOL> return deferred . addCallback ( check ) <EOL> class SimpleAgentProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . made = Deferred ( ) <EOL> self . finished = Deferred ( ) <EOL> self . received = [ ] <EOL> def connectionMade ( self ) : <EOL> self . made . callback ( None ) <EOL> def connectionLost ( self , reason ) : <EOL> self . finished . callback ( None ) <EOL> def dataReceived ( self , data ) : <EOL> self . received . append ( data ) <EOL> class ContentDecoderAgentWithGzipTests ( TestCase , <EOL> FakeReactorAndConnectMixin ) : <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . reactor = self . Reactor ( ) <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> self . agent = client . ContentDecoderAgent ( <EOL> agent , [ ( "<STR_LIT>" , client . GzipDecoder ) ] ) <EOL> def test_gzipEncodingResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> deferred = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> transport = StringTransport ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , transport ) <EOL> response . length = <NUM_LIT:12> <EOL> res . callback ( response ) <EOL> compressor = zlib . compressobj ( <NUM_LIT:2> , zlib . DEFLATED , <NUM_LIT:16> + zlib . MAX_WBITS ) <EOL> data = ( compressor . compress ( '<STR_LIT:x>' * <NUM_LIT:6> ) + compressor . compress ( '<STR_LIT:y>' * <NUM_LIT:4> ) + <EOL> compressor . flush ( ) ) <EOL> def checkResponse ( result ) : <EOL> self . assertNotIdentical ( result , response ) <EOL> self . assertEqual ( result . version , ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) ) <EOL> self . assertEqual ( result . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( result . phrase , '<STR_LIT:OK>' ) <EOL> self . assertEqual ( list ( result . headers . getAllRawHeaders ( ) ) , <EOL> [ ( '<STR_LIT>' , [ '<STR_LIT:bar>' ] ) ] ) <EOL> self . assertEqual ( result . length , UNKNOWN_LENGTH ) <EOL> self . assertRaises ( AttributeError , getattr , result , '<STR_LIT>' ) <EOL> response . _bodyDataReceived ( data [ : <NUM_LIT:5> ] ) <EOL> response . _bodyDataReceived ( data [ <NUM_LIT:5> : ] ) <EOL> response . _bodyDataFinished ( ) <EOL> protocol = SimpleAgentProtocol ( ) <EOL> result . deliverBody ( protocol ) <EOL> self . assertEqual ( protocol . received , [ '<STR_LIT:x>' * <NUM_LIT:6> + '<STR_LIT:y>' * <NUM_LIT:4> ] ) <EOL> return defer . gatherResults ( [ protocol . made , protocol . finished ] ) <EOL> deferred . addCallback ( checkResponse ) <EOL> return deferred <EOL> def test_brokenContent ( self ) : <EOL> """<STR_LIT>""" <EOL> deferred = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> transport = StringTransport ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , transport ) <EOL> response . length = <NUM_LIT:12> <EOL> res . callback ( response ) <EOL> data = "<STR_LIT>" <EOL> def checkResponse ( result ) : <EOL> response . _bodyDataReceived ( data ) <EOL> result . deliverBody ( Protocol ( ) ) <EOL> deferred . addCallback ( checkResponse ) <EOL> self . assertFailure ( deferred , client . ResponseFailed ) <EOL> def checkFailure ( error ) : <EOL> error . reasons [ <NUM_LIT:0> ] . trap ( zlib . error ) <EOL> self . assertIsInstance ( error . response , Response ) <EOL> return deferred . addCallback ( checkFailure ) <EOL> def test_flushData ( self ) : <EOL> """<STR_LIT>""" <EOL> class decompressobj ( object ) : <EOL> def __init__ ( self , wbits ) : <EOL> pass <EOL> def decompress ( self , data ) : <EOL> return '<STR_LIT:x>' <EOL> def flush ( self ) : <EOL> return '<STR_LIT:y>' <EOL> oldDecompressObj = zlib . decompressobj <EOL> zlib . decompressobj = decompressobj <EOL> self . addCleanup ( setattr , zlib , '<STR_LIT>' , oldDecompressObj ) <EOL> deferred = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( { '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> transport = StringTransport ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , transport ) <EOL> res . callback ( response ) <EOL> def checkResponse ( result ) : <EOL> response . _bodyDataReceived ( '<STR_LIT:data>' ) <EOL> response . _bodyDataFinished ( ) <EOL> protocol = SimpleAgentProtocol ( ) <EOL> result . deliverBody ( protocol ) <EOL> self . assertEqual ( protocol . received , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) <EOL> return defer . gatherResults ( [ protocol . made , protocol . finished ] ) <EOL> deferred . addCallback ( checkResponse ) <EOL> return deferred <EOL> def test_flushError ( self ) : <EOL> """<STR_LIT>""" <EOL> class decompressobj ( object ) : <EOL> def __init__ ( self , wbits ) : <EOL> pass <EOL> def decompress ( self , data ) : <EOL> return '<STR_LIT:x>' <EOL> def flush ( self ) : <EOL> raise zlib . error ( ) <EOL> oldDecompressObj = zlib . decompressobj <EOL> zlib . decompressobj = decompressobj <EOL> self . addCleanup ( setattr , zlib , '<STR_LIT>' , oldDecompressObj ) <EOL> deferred = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( { '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> transport = StringTransport ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , transport ) <EOL> res . callback ( response ) <EOL> def checkResponse ( result ) : <EOL> response . _bodyDataReceived ( '<STR_LIT:data>' ) <EOL> response . _bodyDataFinished ( ) <EOL> protocol = SimpleAgentProtocol ( ) <EOL> result . deliverBody ( protocol ) <EOL> self . assertEqual ( protocol . received , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) <EOL> return defer . gatherResults ( [ protocol . made , protocol . finished ] ) <EOL> deferred . addCallback ( checkResponse ) <EOL> self . assertFailure ( deferred , client . ResponseFailed ) <EOL> def checkFailure ( error ) : <EOL> error . reasons [ <NUM_LIT:1> ] . trap ( zlib . error ) <EOL> self . assertIsInstance ( error . response , Response ) <EOL> return deferred . addCallback ( checkFailure ) <EOL> class ProxyAgentTests ( TestCase , FakeReactorAndConnectMixin , AgentTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeAgent ( self ) : <EOL> """<STR_LIT>""" <EOL> return client . ProxyAgent ( <EOL> TCP4ClientEndpoint ( self . reactor , "<STR_LIT:127.0.0.1>" , <NUM_LIT> ) , <EOL> self . reactor ) <EOL> def setUp ( self ) : <EOL> self . reactor = self . Reactor ( ) <EOL> self . agent = client . ProxyAgent ( <EOL> TCP4ClientEndpoint ( self . reactor , "<STR_LIT:bar>" , <NUM_LIT> ) , self . reactor ) <EOL> oldEndpoint = self . agent . _proxyEndpoint <EOL> self . agent . _proxyEndpoint = self . StubEndpoint ( oldEndpoint , self ) <EOL> def test_proxyRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> headers = http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] } ) <EOL> body = object ( ) <EOL> self . agent . request ( <EOL> '<STR_LIT:GET>' , '<STR_LIT>' , headers , body ) <EOL> host , port , factory = self . reactor . tcpClients . pop ( ) [ : <NUM_LIT:3> ] <EOL> self . assertEqual ( host , "<STR_LIT:bar>" ) <EOL> self . assertEqual ( port , <NUM_LIT> ) <EOL> self . assertIsInstance ( factory . _wrappedFactory , <EOL> client . _HTTP11ClientFactory ) <EOL> protocol = self . protocol <EOL> self . assertEqual ( len ( protocol . requests ) , <NUM_LIT:1> ) <EOL> req , res = protocol . requests . pop ( ) <EOL> self . assertIsInstance ( req , Request ) <EOL> self . assertEqual ( req . method , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( req . uri , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> req . headers , <EOL> http_headers . Headers ( { '<STR_LIT:foo>' : [ '<STR_LIT:bar>' ] , <EOL> '<STR_LIT:host>' : [ '<STR_LIT>' ] } ) ) <EOL> self . assertIdentical ( req . bodyProducer , body ) <EOL> def test_nonPersistent ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . agent . _pool . persistent , False ) <EOL> def test_connectUsesConnectionPool ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = DummyEndpoint ( ) <EOL> class DummyPool ( object ) : <EOL> connected = False <EOL> persistent = False <EOL> def getConnection ( this , key , ep ) : <EOL> this . connected = True <EOL> self . assertIdentical ( ep , endpoint ) <EOL> self . assertEqual ( key , ( "<STR_LIT>" , endpoint ) ) <EOL> return defer . succeed ( StubHTTPProtocol ( ) ) <EOL> pool = DummyPool ( ) <EOL> agent = client . ProxyAgent ( endpoint , self . reactor , pool = pool ) <EOL> self . assertIdentical ( pool , agent . _pool ) <EOL> agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( agent . _pool . connected , True ) <EOL> class _RedirectAgentTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def test_noRedirect ( self ) : <EOL> """<STR_LIT>""" <EOL> deferred = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . protocol . requests ) ) <EOL> result = self . successResultOf ( deferred ) <EOL> self . assertIdentical ( response , result ) <EOL> self . assertIdentical ( result . previousResponse , None ) <EOL> def _testRedirectDefault ( self , code ) : <EOL> """<STR_LIT>""" <EOL> self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> host , port = self . reactor . tcpClients . pop ( ) [ : <NUM_LIT:2> ] <EOL> self . assertEqual ( "<STR_LIT>" , host ) <EOL> self . assertEqual ( <NUM_LIT> , port ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( <EOL> { '<STR_LIT:location>' : [ '<STR_LIT>' ] } ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , code , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> req2 , res2 = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( '<STR_LIT:GET>' , req2 . method ) <EOL> self . assertEqual ( '<STR_LIT>' , req2 . uri ) <EOL> host , port = self . reactor . sslClients . pop ( ) [ : <NUM_LIT:2> ] <EOL> self . assertEqual ( "<STR_LIT>" , host ) <EOL> self . assertEqual ( <NUM_LIT> , port ) <EOL> def test_redirect301 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectDefault ( <NUM_LIT> ) <EOL> def test_redirect302 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectDefault ( <NUM_LIT> ) <EOL> def test_redirect307 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectDefault ( <NUM_LIT> ) <EOL> def _testRedirectToGet ( self , code , method ) : <EOL> """<STR_LIT>""" <EOL> self . agent . request ( method , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( <EOL> { '<STR_LIT:location>' : [ '<STR_LIT>' ] } ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , code , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> req2 , res2 = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( '<STR_LIT:GET>' , req2 . method ) <EOL> self . assertEqual ( '<STR_LIT>' , req2 . uri ) <EOL> def test_redirect303 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectToGet ( <NUM_LIT> , '<STR_LIT:POST>' ) <EOL> def test_noLocationField ( self ) : <EOL> """<STR_LIT>""" <EOL> deferred = self . agent . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT> , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> fail = self . failureResultOf ( deferred , client . ResponseFailed ) <EOL> fail . value . reasons [ <NUM_LIT:0> ] . trap ( error . RedirectWithNoLocation ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> fail . value . reasons [ <NUM_LIT:0> ] . value . uri ) <EOL> self . assertEqual ( <NUM_LIT> , fail . value . response . code ) <EOL> def _testPageRedirectFailure ( self , code , method ) : <EOL> """<STR_LIT>""" <EOL> deferred = self . agent . request ( method , '<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , code , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> fail = self . failureResultOf ( deferred , client . ResponseFailed ) <EOL> fail . value . reasons [ <NUM_LIT:0> ] . trap ( error . PageRedirect ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> fail . value . reasons [ <NUM_LIT:0> ] . value . location ) <EOL> self . assertEqual ( code , fail . value . response . code ) <EOL> def test_307OnPost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testPageRedirectFailure ( <NUM_LIT> , '<STR_LIT:POST>' ) <EOL> def test_redirectLimit ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> redirectAgent = client . RedirectAgent ( agent , <NUM_LIT:1> ) <EOL> deferred = redirectAgent . request ( b'<STR_LIT:GET>' , b'<STR_LIT>' ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( <EOL> { b'<STR_LIT:location>' : [ b'<STR_LIT>' ] } ) <EOL> response = Response ( ( b'<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT> , b'<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> req2 , res2 = self . protocol . requests . pop ( ) <EOL> response2 = Response ( ( b'<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT> , b'<STR_LIT:OK>' , headers , None ) <EOL> res2 . callback ( response2 ) <EOL> fail = self . failureResultOf ( deferred , client . ResponseFailed ) <EOL> fail . value . reasons [ <NUM_LIT:0> ] . trap ( error . InfiniteRedirection ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> fail . value . reasons [ <NUM_LIT:0> ] . value . location ) <EOL> self . assertEqual ( <NUM_LIT> , fail . value . response . code ) <EOL> def _testRedirectURI ( self , uri , location , finalURI ) : <EOL> """<STR_LIT>""" <EOL> self . agent . request ( '<STR_LIT:GET>' , uri ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( <EOL> { '<STR_LIT:location>' : [ location ] } ) <EOL> response = Response ( ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT> , '<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> req2 , res2 = self . protocol . requests . pop ( ) <EOL> self . assertEqual ( '<STR_LIT:GET>' , req2 . method ) <EOL> self . assertEqual ( finalURI , req2 . absoluteURI ) <EOL> def test_relativeURI ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_relativeURIPreserveFragments ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_relativeURISchemeRelative ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . _testRedirectURI ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_responseHistory ( self ) : <EOL> """<STR_LIT>""" <EOL> agent = self . buildAgentForWrapperTest ( self . reactor ) <EOL> redirectAgent = client . RedirectAgent ( agent ) <EOL> deferred = redirectAgent . request ( b'<STR_LIT:GET>' , b'<STR_LIT>' ) <EOL> redirectReq , redirectRes = self . protocol . requests . pop ( ) <EOL> headers = http_headers . Headers ( <EOL> { b'<STR_LIT:location>' : [ b'<STR_LIT>' ] } ) <EOL> redirectResponse = Response ( ( b'<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT> , b'<STR_LIT:OK>' , headers , None ) <EOL> redirectRes . callback ( redirectResponse ) <EOL> req , res = self . protocol . requests . pop ( ) <EOL> response = Response ( ( b'<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) , <NUM_LIT:200> , b'<STR_LIT:OK>' , headers , None ) <EOL> res . callback ( response ) <EOL> finalResponse = self . successResultOf ( deferred ) <EOL> self . assertIdentical ( finalResponse . previousResponse , redirectResponse ) <EOL> self . assertIdentical ( redirectResponse . previousResponse , None ) <EOL> class RedirectAgentTests ( TestCase , FakeReactorAndConnectMixin , <EOL> _RedirectAgentTestsMixin , AgentTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeAgent ( self ) : <EOL> """<STR_LIT>""" <EOL> return client . RedirectAgent ( <EOL> self . buildAgentForWrapperTest ( self . reactor ) ) <EOL> def setUp ( self ) : <EOL> self . reactor = self . Reactor ( ) <EOL> self . agent = self . makeAgent ( ) <EOL> def test_301OnPost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testPageRedirectFailure ( <NUM_LIT> , '<STR_LIT:POST>' ) <EOL> def test_302OnPost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testPageRedirectFailure ( <NUM_LIT> , '<STR_LIT:POST>' ) <EOL> class BrowserLikeRedirectAgentTests ( TestCase , <EOL> FakeReactorAndConnectMixin , <EOL> _RedirectAgentTestsMixin , <EOL> AgentTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def makeAgent ( self ) : <EOL> """<STR_LIT>""" <EOL> return client . BrowserLikeRedirectAgent ( <EOL> self . buildAgentForWrapperTest ( self . reactor ) ) <EOL> def setUp ( self ) : <EOL> self . reactor = self . Reactor ( ) <EOL> self . agent = self . makeAgent ( ) <EOL> def test_redirectToGet301 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectToGet ( <NUM_LIT> , '<STR_LIT:POST>' ) <EOL> def test_redirectToGet302 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testRedirectToGet ( <NUM_LIT> , '<STR_LIT:POST>' ) <EOL> class AbortableStringTransport ( StringTransport ) : <EOL> """<STR_LIT>""" <EOL> aborting = False <EOL> def abortConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . aborting = True <EOL> self . loseConnection ( ) <EOL> class DummyResponse ( object ) : <EOL> """<STR_LIT>""" <EOL> code = <NUM_LIT:200> <EOL> phrase = "<STR_LIT:OK>" <EOL> def __init__ ( self , headers = None , transportFactory = AbortableStringTransport ) : <EOL> """<STR_LIT>""" <EOL> if headers is None : <EOL> headers = Headers ( ) <EOL> self . headers = headers <EOL> self . transport = transportFactory ( ) <EOL> def deliverBody ( self , protocol ) : <EOL> """<STR_LIT>""" <EOL> self . protocol = protocol <EOL> self . protocol . makeConnection ( self . transport ) <EOL> class ReadBodyTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_success ( self ) : <EOL> """<STR_LIT>""" <EOL> response = DummyResponse ( ) <EOL> d = client . readBody ( response ) <EOL> response . protocol . dataReceived ( "<STR_LIT>" ) <EOL> response . protocol . dataReceived ( "<STR_LIT>" ) <EOL> response . protocol . connectionLost ( Failure ( ResponseDone ( ) ) ) <EOL> self . assertEqual ( self . successResultOf ( d ) , "<STR_LIT>" ) <EOL> def test_cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> response = DummyResponse ( ) <EOL> deferred = client . readBody ( response ) <EOL> deferred . cancel ( ) <EOL> self . failureResultOf ( deferred , defer . CancelledError ) <EOL> self . assertTrue ( response . transport . aborting ) <EOL> def test_withPotentialDataLoss ( self ) : <EOL> """<STR_LIT>""" <EOL> response = DummyResponse ( ) <EOL> d = client . readBody ( response ) <EOL> response . protocol . dataReceived ( "<STR_LIT>" ) <EOL> response . protocol . dataReceived ( "<STR_LIT>" ) <EOL> response . protocol . connectionLost ( Failure ( PotentialDataLoss ( ) ) ) <EOL> failure = self . failureResultOf ( d ) <EOL> failure . trap ( client . PartialDownloadError ) <EOL> self . assertEqual ( { <EOL> "<STR_LIT:status>" : failure . value . status , <EOL> "<STR_LIT:message>" : failure . value . message , <EOL> "<STR_LIT:body>" : failure . value . response , <EOL> } , { <EOL> "<STR_LIT:status>" : <NUM_LIT:200> , <EOL> "<STR_LIT:message>" : "<STR_LIT:OK>" , <EOL> "<STR_LIT:body>" : "<STR_LIT>" , <EOL> } ) <EOL> def test_otherErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> response = DummyResponse ( ) <EOL> d = client . readBody ( response ) <EOL> response . protocol . dataReceived ( "<STR_LIT>" ) <EOL> response . protocol . connectionLost ( <EOL> Failure ( ConnectionLost ( "<STR_LIT>" ) ) ) <EOL> reason = self . failureResultOf ( d ) <EOL> reason . trap ( ConnectionLost ) <EOL> self . assertEqual ( reason . value . args , ( "<STR_LIT>" , ) ) <EOL> def test_deprecatedTransport ( self ) : <EOL> """<STR_LIT>""" <EOL> response = DummyResponse ( transportFactory = StringTransport ) <EOL> d = self . assertWarns ( <EOL> DeprecationWarning , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> __file__ , <EOL> lambda : client . readBody ( response ) ) <EOL> d . cancel ( ) <EOL> self . failureResultOf ( d , defer . CancelledError ) </s>
