<s> """<STR_LIT>""" <EOL> from cStringIO import StringIO <EOL> from zope . interface import implements <EOL> from zope . interface . verify import verifyObject <EOL> from twisted . trial import unittest <EOL> from twisted . internet import reactor <EOL> from twisted . internet . address import IPv4Address <EOL> from twisted . internet . defer import Deferred <EOL> from twisted . web import server , resource , util <EOL> from twisted . internet import defer , interfaces , task <EOL> from twisted . web import iweb , http , http_headers , error <EOL> from twisted . python import log <EOL> class DummyRequest : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' <EOL> method = '<STR_LIT:GET>' <EOL> client = None <EOL> def registerProducer ( self , prod , s ) : <EOL> self . go = <NUM_LIT:1> <EOL> while self . go : <EOL> prod . resumeProducing ( ) <EOL> def unregisterProducer ( self ) : <EOL> self . go = <NUM_LIT:0> <EOL> def __init__ ( self , postpath , session = None ) : <EOL> self . sitepath = [ ] <EOL> self . written = [ ] <EOL> self . finished = <NUM_LIT:0> <EOL> self . postpath = postpath <EOL> self . prepath = [ ] <EOL> self . session = None <EOL> self . protoSession = session or server . Session ( <NUM_LIT:0> , self ) <EOL> self . args = { } <EOL> self . outgoingHeaders = { } <EOL> self . responseHeaders = http_headers . Headers ( ) <EOL> self . responseCode = None <EOL> self . headers = { } <EOL> self . _finishedDeferreds = [ ] <EOL> def getHeader ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . headers . get ( name . lower ( ) , None ) <EOL> def setHeader ( self , name , value ) : <EOL> """<STR_LIT>""" <EOL> self . outgoingHeaders [ name . lower ( ) ] = value <EOL> def getSession ( self ) : <EOL> if self . session : <EOL> return self . session <EOL> assert not self . written , "<STR_LIT>" <EOL> self . session = self . protoSession <EOL> return self . session <EOL> def render ( self , resource ) : <EOL> """<STR_LIT>""" <EOL> result = resource . render ( self ) <EOL> if result is server . NOT_DONE_YET : <EOL> return <EOL> self . write ( result ) <EOL> self . finish ( ) <EOL> def write ( self , data ) : <EOL> self . written . append ( data ) <EOL> def notifyFinish ( self ) : <EOL> """<STR_LIT>""" <EOL> finished = Deferred ( ) <EOL> self . _finishedDeferreds . append ( finished ) <EOL> return finished <EOL> def finish ( self ) : <EOL> """<STR_LIT>""" <EOL> self . finished = self . finished + <NUM_LIT:1> <EOL> if self . _finishedDeferreds is not None : <EOL> observers = self . _finishedDeferreds <EOL> self . _finishedDeferreds = None <EOL> for obs in observers : <EOL> obs . callback ( None ) <EOL> def processingFailed ( self , reason ) : <EOL> """<STR_LIT>""" <EOL> if self . _finishedDeferreds is not None : <EOL> observers = self . _finishedDeferreds <EOL> self . _finishedDeferreds = None <EOL> for obs in observers : <EOL> obs . errback ( reason ) <EOL> def addArg ( self , name , value ) : <EOL> self . args [ name ] = [ value ] <EOL> def setResponseCode ( self , code , message = None ) : <EOL> """<STR_LIT>""" <EOL> assert not self . written , "<STR_LIT>" % "<STR_LIT>" . join ( self . written ) <EOL> self . responseCode = code <EOL> self . responseMessage = message <EOL> def setLastModified ( self , when ) : <EOL> assert not self . written , "<STR_LIT>" % "<STR_LIT>" . join ( self . written ) <EOL> def setETag ( self , tag ) : <EOL> assert not self . written , "<STR_LIT>" % "<STR_LIT>" . join ( self . written ) <EOL> def getClientIP ( self ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( self . client , IPv4Address ) : <EOL> return self . client . host <EOL> return None <EOL> class ResourceTestCase ( unittest . TestCase ) : <EOL> def testListEntities ( self ) : <EOL> r = resource . Resource ( ) <EOL> self . assertEqual ( [ ] , r . listEntities ( ) ) <EOL> class SimpleResource ( resource . Resource ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , contentType = None ) : <EOL> resource . Resource . __init__ ( self ) <EOL> self . _contentType = contentType <EOL> def render ( self , request ) : <EOL> if self . _contentType is not None : <EOL> request . responseHeaders . setRawHeaders ( <EOL> "<STR_LIT>" , [ self . _contentType ] ) <EOL> if http . CACHED in ( request . setLastModified ( <NUM_LIT:10> ) , <EOL> request . setETag ( '<STR_LIT>' ) ) : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return "<STR_LIT>" <EOL> class DummyChannel : <EOL> class TCP : <EOL> port = <NUM_LIT> <EOL> disconnected = False <EOL> def __init__ ( self ) : <EOL> self . written = StringIO ( ) <EOL> self . producers = [ ] <EOL> def getPeer ( self ) : <EOL> return IPv4Address ( "<STR_LIT>" , '<STR_LIT>' , <NUM_LIT> ) <EOL> def write ( self , bytes ) : <EOL> assert isinstance ( bytes , str ) <EOL> self . written . write ( bytes ) <EOL> def writeSequence ( self , iovec ) : <EOL> map ( self . write , iovec ) <EOL> def getHost ( self ) : <EOL> return IPv4Address ( "<STR_LIT>" , '<STR_LIT>' , self . port ) <EOL> def registerProducer ( self , producer , streaming ) : <EOL> self . producers . append ( ( producer , streaming ) ) <EOL> def loseConnection ( self ) : <EOL> self . disconnected = True <EOL> class SSL ( TCP ) : <EOL> implements ( interfaces . ISSLTransport ) <EOL> site = server . Site ( resource . Resource ( ) ) <EOL> def __init__ ( self ) : <EOL> self . transport = self . TCP ( ) <EOL> def requestDone ( self , request ) : <EOL> pass <EOL> class SiteTest ( unittest . TestCase ) : <EOL> def test_simplestSite ( self ) : <EOL> """<STR_LIT>""" <EOL> sres1 = SimpleResource ( ) <EOL> sres2 = SimpleResource ( ) <EOL> sres1 . putChild ( "<STR_LIT>" , sres2 ) <EOL> site = server . Site ( sres1 ) <EOL> self . assertIdentical ( <EOL> site . getResourceFor ( DummyRequest ( [ '<STR_LIT>' ] ) ) , <EOL> sres2 , "<STR_LIT>" ) <EOL> class SessionTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . clock = task . Clock ( ) <EOL> self . uid = '<STR_LIT>' <EOL> self . site = server . Site ( resource . Resource ( ) ) <EOL> self . session = server . Session ( self . site , self . uid , self . clock ) <EOL> self . site . sessions [ self . uid ] = self . session <EOL> def test_defaultReactor ( self ) : <EOL> """<STR_LIT>""" <EOL> session = server . Session ( server . Site ( resource . Resource ( ) ) , '<STR_LIT>' ) <EOL> self . assertIdentical ( session . _reactor , reactor ) <EOL> def test_startCheckingExpiration ( self ) : <EOL> """<STR_LIT>""" <EOL> self . session . startCheckingExpiration ( ) <EOL> self . clock . advance ( self . session . sessionTimeout - <NUM_LIT:1> ) <EOL> self . assertIn ( self . uid , self . site . sessions ) <EOL> self . clock . advance ( <NUM_LIT:1> ) <EOL> self . assertNotIn ( self . uid , self . site . sessions ) <EOL> self . assertFalse ( self . clock . calls ) <EOL> def test_expire ( self ) : <EOL> """<STR_LIT>""" <EOL> self . session . expire ( ) <EOL> self . assertNotIn ( self . uid , self . site . sessions ) <EOL> self . assertFalse ( self . clock . calls ) <EOL> def test_expireWhileChecking ( self ) : <EOL> """<STR_LIT>""" <EOL> self . session . startCheckingExpiration ( ) <EOL> self . test_expire ( ) <EOL> def test_notifyOnExpire ( self ) : <EOL> """<STR_LIT>""" <EOL> callbackRan = [ False ] <EOL> def expired ( ) : <EOL> callbackRan [ <NUM_LIT:0> ] = True <EOL> self . session . notifyOnExpire ( expired ) <EOL> self . session . expire ( ) <EOL> self . assertTrue ( callbackRan [ <NUM_LIT:0> ] ) <EOL> def test_touch ( self ) : <EOL> """<STR_LIT>""" <EOL> self . clock . advance ( <NUM_LIT:3> ) <EOL> self . session . touch ( ) <EOL> self . assertEqual ( self . session . lastModified , <NUM_LIT:3> ) <EOL> self . session . startCheckingExpiration ( ) <EOL> self . clock . advance ( self . session . sessionTimeout - <NUM_LIT:1> ) <EOL> self . session . touch ( ) <EOL> self . clock . advance ( self . session . sessionTimeout - <NUM_LIT:1> ) <EOL> self . assertIn ( self . uid , self . site . sessions ) <EOL> self . clock . advance ( <NUM_LIT:1> ) <EOL> self . assertNotIn ( self . uid , self . site . sessions ) <EOL> def test_startCheckingExpirationParameterDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> self . session . startCheckingExpiration ( <NUM_LIT> ) <EOL> warnings = self . flushWarnings ( [ <EOL> self . test_startCheckingExpirationParameterDeprecated ] ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_checkExpiredDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> self . session . checkExpired ( ) <EOL> warnings = self . flushWarnings ( [ self . test_checkExpiredDeprecated ] ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> def httpBody ( whole ) : <EOL> return whole . split ( '<STR_LIT>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> def httpHeader ( whole , key ) : <EOL> key = key . lower ( ) <EOL> headers = whole . split ( '<STR_LIT>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> for header in headers . split ( '<STR_LIT:\r\n>' ) : <EOL> if header . lower ( ) . startswith ( key ) : <EOL> return header . split ( '<STR_LIT::>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] . strip ( ) <EOL> return None <EOL> def httpCode ( whole ) : <EOL> l1 = whole . split ( '<STR_LIT:\r\n>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> return int ( l1 . split ( ) [ <NUM_LIT:1> ] ) <EOL> class ConditionalTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . resrc = SimpleResource ( ) <EOL> self . resrc . putChild ( '<STR_LIT>' , self . resrc ) <EOL> self . resrc . putChild ( '<STR_LIT>' , SimpleResource ( '<STR_LIT>' ) ) <EOL> self . site = server . Site ( self . resrc ) <EOL> self . site . logFile = log . logfile <EOL> self . channel = self . site . buildProtocol ( None ) <EOL> self . transport = http . StringTransport ( ) <EOL> self . transport . close = lambda * a , ** kw : None <EOL> self . transport . disconnecting = lambda * a , ** kw : <NUM_LIT:0> <EOL> self . transport . getPeer = lambda * a , ** kw : "<STR_LIT>" <EOL> self . transport . getHost = lambda * a , ** kw : "<STR_LIT:host>" <EOL> self . channel . makeConnection ( self . transport ) <EOL> def tearDown ( self ) : <EOL> self . channel . connectionLost ( None ) <EOL> def _modifiedTest ( self , modifiedSince = None , etag = None ) : <EOL> """<STR_LIT>""" <EOL> if modifiedSince is not None : <EOL> validator = "<STR_LIT>" + modifiedSince <EOL> else : <EOL> validator = "<STR_LIT>" + etag <EOL> for line in [ "<STR_LIT>" , validator , "<STR_LIT>" ] : <EOL> self . channel . lineReceived ( line ) <EOL> result = self . transport . getvalue ( ) <EOL> self . assertEqual ( httpCode ( result ) , http . OK ) <EOL> self . assertEqual ( httpBody ( result ) , "<STR_LIT>" ) <EOL> self . assertEqual ( httpHeader ( result , "<STR_LIT:Content-Type>" ) , "<STR_LIT>" ) <EOL> def test_modified ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _modifiedTest ( modifiedSince = http . datetimeToString ( <NUM_LIT:1> ) ) <EOL> def test_unmodified ( self ) : <EOL> """<STR_LIT>""" <EOL> for line in [ "<STR_LIT>" , <EOL> "<STR_LIT>" + http . datetimeToString ( <NUM_LIT:100> ) , "<STR_LIT>" ] : <EOL> self . channel . lineReceived ( line ) <EOL> result = self . transport . getvalue ( ) <EOL> self . assertEqual ( httpCode ( result ) , http . NOT_MODIFIED ) <EOL> self . assertEqual ( httpBody ( result ) , "<STR_LIT>" ) <EOL> self . assertEqual ( httpHeader ( result , "<STR_LIT:Content-Type>" ) , None ) <EOL> def test_invalidTimestamp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _modifiedTest ( modifiedSince = "<STR_LIT>" ) <EOL> def test_invalidTimestampYear ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _modifiedTest ( modifiedSince = "<STR_LIT>" ) <EOL> def test_invalidTimestampTooLongAgo ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _modifiedTest ( modifiedSince = "<STR_LIT>" ) <EOL> def test_invalidTimestampMonth ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _modifiedTest ( modifiedSince = "<STR_LIT>" ) <EOL> def test_etagMatchedNot ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _modifiedTest ( etag = "<STR_LIT>" ) <EOL> def test_etagMatched ( self ) : <EOL> """<STR_LIT>""" <EOL> for line in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . channel . lineReceived ( line ) <EOL> result = self . transport . getvalue ( ) <EOL> self . assertEqual ( httpHeader ( result , "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( httpCode ( result ) , http . NOT_MODIFIED ) <EOL> self . assertEqual ( httpBody ( result ) , "<STR_LIT>" ) <EOL> def test_unmodifiedWithContentType ( self ) : <EOL> """<STR_LIT>""" <EOL> for line in [ "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . channel . lineReceived ( line ) <EOL> result = self . transport . getvalue ( ) <EOL> self . assertEqual ( httpCode ( result ) , http . NOT_MODIFIED ) <EOL> self . assertEqual ( httpBody ( result ) , "<STR_LIT>" ) <EOL> self . assertEqual ( httpHeader ( result , "<STR_LIT:Content-Type>" ) , "<STR_LIT>" ) <EOL> class RequestTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( <EOL> verifyObject ( iweb . IRequest , server . Request ( DummyChannel ( ) , True ) ) ) <EOL> def testChildLink ( self ) : <EOL> request = server . Request ( DummyChannel ( ) , <NUM_LIT:1> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . childLink ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> request = server . Request ( DummyChannel ( ) , <NUM_LIT:1> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . childLink ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def testPrePathURLSimple ( self ) : <EOL> request = server . Request ( DummyChannel ( ) , <NUM_LIT:1> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def testPrePathURLNonDefault ( self ) : <EOL> d = DummyChannel ( ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def testPrePathURLSSLPort ( self ) : <EOL> d = DummyChannel ( ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def testPrePathURLSSLPortAndSSL ( self ) : <EOL> d = DummyChannel ( ) <EOL> d . transport = DummyChannel . SSL ( ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def testPrePathURLHTTPPortAndSSL ( self ) : <EOL> d = DummyChannel ( ) <EOL> d . transport = DummyChannel . SSL ( ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def testPrePathURLSSLNonDefault ( self ) : <EOL> d = DummyChannel ( ) <EOL> d . transport = DummyChannel . SSL ( ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def testPrePathURLSetSSLHost ( self ) : <EOL> d = DummyChannel ( ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> def test_prePathURLQuoting ( self ) : <EOL> """<STR_LIT>""" <EOL> d = DummyChannel ( ) <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( request . prePathURL ( ) , '<STR_LIT>' ) <EOL> class RootResource ( resource . Resource ) : <EOL> isLeaf = <NUM_LIT:0> <EOL> def getChildWithDefault ( self , name , request ) : <EOL> request . rememberRootURL ( ) <EOL> return resource . Resource . getChildWithDefault ( self , name , request ) <EOL> def render ( self , request ) : <EOL> return '<STR_LIT>' <EOL> class RememberURLTest ( unittest . TestCase ) : <EOL> def createServer ( self , r ) : <EOL> chan = DummyChannel ( ) <EOL> chan . site = server . Site ( r ) <EOL> return chan <EOL> def testSimple ( self ) : <EOL> r = resource . Resource ( ) <EOL> r . isLeaf = <NUM_LIT:0> <EOL> rr = RootResource ( ) <EOL> r . putChild ( '<STR_LIT:foo>' , rr ) <EOL> rr . putChild ( '<STR_LIT>' , rr ) <EOL> rr . putChild ( '<STR_LIT:bar>' , resource . Resource ( ) ) <EOL> chan = self . createServer ( r ) <EOL> for url in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> request = server . Request ( chan , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , url , '<STR_LIT>' ) <EOL> self . assertEqual ( request . getRootURL ( ) , "<STR_LIT>" ) <EOL> def testRoot ( self ) : <EOL> rr = RootResource ( ) <EOL> rr . putChild ( '<STR_LIT>' , rr ) <EOL> rr . putChild ( '<STR_LIT:bar>' , resource . Resource ( ) ) <EOL> chan = self . createServer ( rr ) <EOL> for url in [ '<STR_LIT:/>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> request = server . Request ( chan , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> request . requestReceived ( '<STR_LIT:GET>' , url , '<STR_LIT>' ) <EOL> self . assertEqual ( request . getRootURL ( ) , "<STR_LIT>" ) <EOL> class NewRenderResource ( resource . Resource ) : <EOL> def render_GET ( self , request ) : <EOL> return "<STR_LIT>" <EOL> def render_HEH ( self , request ) : <EOL> return "<STR_LIT>" <EOL> class HeadlessResource ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( resource . IResource ) <EOL> allowedMethods = [ "<STR_LIT:GET>" ] <EOL> def render ( self , request ) : <EOL> """<STR_LIT>""" <EOL> self . request = request <EOL> if request . method not in self . allowedMethods : <EOL> raise error . UnsupportedMethod ( self . allowedMethods ) <EOL> self . request . write ( "<STR_LIT>" ) <EOL> return server . NOT_DONE_YET <EOL> class NewRenderTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _getReq ( self , resource = None ) : <EOL> """<STR_LIT>""" <EOL> d = DummyChannel ( ) <EOL> if resource is None : <EOL> resource = NewRenderResource ( ) <EOL> d . site . resource . putChild ( '<STR_LIT>' , resource ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> return request <EOL> def testGoodMethods ( self ) : <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . transport . getvalue ( ) . splitlines ( ) [ - <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . transport . getvalue ( ) . splitlines ( ) [ - <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> def testBadMethods ( self ) : <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT> ) <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT> ) <EOL> def testImplicitHead ( self ) : <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( - <NUM_LIT:1> , req . transport . getvalue ( ) . find ( '<STR_LIT>' ) ) <EOL> def test_unsupportedHead ( self ) : <EOL> """<STR_LIT>""" <EOL> resource = HeadlessResource ( ) <EOL> req = self . _getReq ( resource ) <EOL> req . requestReceived ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> headers , body = req . transport . getvalue ( ) . split ( '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( body , '<STR_LIT>' ) <EOL> class GettableResource ( resource . Resource ) : <EOL> """<STR_LIT>""" <EOL> def render_GET ( self ) : <EOL> pass <EOL> def render_fred_render_ethel ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class AllowedMethodsTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _getReq ( self ) : <EOL> """<STR_LIT>""" <EOL> d = DummyChannel ( ) <EOL> d . site . resource . putChild ( '<STR_LIT>' , GettableResource ( ) ) <EOL> d . transport . port = <NUM_LIT> <EOL> request = server . Request ( d , <NUM_LIT:1> ) <EOL> request . setHost ( '<STR_LIT>' , <NUM_LIT> ) <EOL> request . gotLength ( <NUM_LIT:0> ) <EOL> return request <EOL> def test_computeAllowedMethods ( self ) : <EOL> """<STR_LIT>""" <EOL> res = GettableResource ( ) <EOL> allowedMethods = resource . _computeAllowedMethods ( res ) <EOL> self . assertEqual ( set ( allowedMethods ) , <EOL> set ( [ '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> def test_notAllowed ( self ) : <EOL> """<STR_LIT>""" <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT:POST>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> set ( req . responseHeaders . getRawHeaders ( '<STR_LIT>' ) [ <NUM_LIT:0> ] . split ( "<STR_LIT:U+002CU+0020>" ) ) , <EOL> set ( [ '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> ) <EOL> def test_notAllowedQuoting ( self ) : <EOL> """<STR_LIT>""" <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT:POST>' , '<STR_LIT>' <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT> ) <EOL> renderedPage = req . transport . getvalue ( ) <EOL> self . assertNotIn ( "<STR_LIT>" , renderedPage ) <EOL> self . assertIn ( '<STR_LIT>' , renderedPage ) <EOL> def test_notImplementedQuoting ( self ) : <EOL> """<STR_LIT>""" <EOL> req = self . _getReq ( ) <EOL> req . requestReceived ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( req . code , <NUM_LIT> ) <EOL> renderedPage = req . transport . getvalue ( ) <EOL> self . assertNotIn ( "<STR_LIT>" , renderedPage ) <EOL> self . assertIn ( '<STR_LIT>' , renderedPage ) <EOL> class SDResource ( resource . Resource ) : <EOL> def __init__ ( self , default ) : <EOL> self . default = default <EOL> def getChildWithDefault ( self , name , request ) : <EOL> d = defer . succeed ( self . default ) <EOL> resource = util . DeferredResource ( d ) <EOL> return resource . getChildWithDefault ( name , request ) <EOL> class DeferredResourceTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testDeferredResource ( self ) : <EOL> r = resource . Resource ( ) <EOL> r . isLeaf = <NUM_LIT:1> <EOL> s = SDResource ( r ) <EOL> d = DummyRequest ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ] ) <EOL> resource . getChildForRequest ( s , d ) <EOL> self . assertEqual ( d . postpath , [ '<STR_LIT:bar>' , '<STR_LIT>' ] ) <EOL> def test_render ( self ) : <EOL> """<STR_LIT>""" <EOL> rendered = [ ] <EOL> request = DummyRequest ( [ ] ) <EOL> request . render = rendered . append <EOL> result = resource . Resource ( ) <EOL> deferredResource = util . DeferredResource ( defer . succeed ( result ) ) <EOL> deferredResource . render ( request ) <EOL> self . assertEqual ( rendered , [ result ] ) <EOL> class DummyRequestForLogTest ( DummyRequest ) : <EOL> uri = '<STR_LIT>' <EOL> code = <NUM_LIT> <EOL> clientproto = '<STR_LIT>' <EOL> sentLength = None <EOL> client = IPv4Address ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) <EOL> class TestLogEscaping ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . site = http . HTTPFactory ( ) <EOL> self . site . logFile = StringIO ( ) <EOL> self . request = DummyRequestForLogTest ( self . site , False ) <EOL> def testSimple ( self ) : <EOL> self . site . _logDateTime = "<STR_LIT>" % ( <EOL> <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> ) <EOL> self . site . log ( self . request ) <EOL> self . site . logFile . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <EOL> self . site . logFile . read ( ) , <EOL> '<STR_LIT>' ) <EOL> def testMethodQuote ( self ) : <EOL> self . site . _logDateTime = "<STR_LIT>" % ( <EOL> <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> ) <EOL> self . request . method = '<STR_LIT>' <EOL> self . site . log ( self . request ) <EOL> self . site . logFile . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <EOL> self . site . logFile . read ( ) , <EOL> '<STR_LIT>' ) <EOL> def testRequestQuote ( self ) : <EOL> self . site . _logDateTime = "<STR_LIT>" % ( <EOL> <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> ) <EOL> self . request . uri = '<STR_LIT>' <EOL> self . site . log ( self . request ) <EOL> self . site . logFile . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <EOL> self . site . logFile . read ( ) , <EOL> '<STR_LIT>' ) <EOL> def testProtoQuote ( self ) : <EOL> self . site . _logDateTime = "<STR_LIT>" % ( <EOL> <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> ) <EOL> self . request . clientproto = '<STR_LIT>' <EOL> self . site . log ( self . request ) <EOL> self . site . logFile . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <EOL> self . site . logFile . read ( ) , <EOL> '<STR_LIT>' ) <EOL> def testRefererQuote ( self ) : <EOL> self . site . _logDateTime = "<STR_LIT>" % ( <EOL> <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> ) <EOL> self . request . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . site . log ( self . request ) <EOL> self . site . logFile . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <EOL> self . site . logFile . read ( ) , <EOL> '<STR_LIT>' ) <EOL> def testUserAgentQuote ( self ) : <EOL> self . site . _logDateTime = "<STR_LIT>" % ( <EOL> <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> ) <EOL> self . request . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . site . log ( self . request ) <EOL> self . site . logFile . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <EOL> self . site . logFile . read ( ) , <EOL> '<STR_LIT>' ) <EOL> class ServerAttributesTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_deprecatedAttributeDateTimeString ( self ) : <EOL> """<STR_LIT>""" <EOL> deprecated_func = server . date_time_string <EOL> warnings = self . flushWarnings ( <EOL> offendingFunctions = [ self . test_deprecatedAttributeDateTimeString ] ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_deprecatedAttributeStringDateTime ( self ) : <EOL> """<STR_LIT>""" <EOL> deprecated_func = server . string_date_time <EOL> warnings = self . flushWarnings ( <EOL> offendingFunctions = [ self . test_deprecatedAttributeStringDateTime ] ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) </s>
