<s> """<STR_LIT>""" <EOL> from os . path import abspath <EOL> from xml . dom . minidom import parseString <EOL> try : <EOL> import pwd <EOL> except ImportError : <EOL> pwd = None <EOL> from zope . interface . verify import verifyObject <EOL> from twisted . python import log , filepath <EOL> from twisted . internet import reactor , defer <EOL> from twisted . trial import unittest <EOL> from twisted . spread import pb <EOL> from twisted . spread . banana import SIZE_LIMIT <EOL> from twisted . web import http , distrib , client , resource , static , server <EOL> from twisted . web . test . test_web import DummyRequest <EOL> from twisted . web . test . _util import _render <EOL> from twisted . test import proto_helpers <EOL> class MySite ( server . Site ) : <EOL> pass <EOL> class PBServerFactory ( pb . PBServerFactory ) : <EOL> """<STR_LIT>""" <EOL> proto = None <EOL> def buildProtocol ( self , addr ) : <EOL> self . proto = pb . PBServerFactory . buildProtocol ( self , addr ) <EOL> return self . proto <EOL> class DistribTest ( unittest . TestCase ) : <EOL> port1 = None <EOL> port2 = None <EOL> sub = None <EOL> f1 = None <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> dl = [ defer . Deferred ( ) , defer . Deferred ( ) ] <EOL> if self . f1 is not None and self . f1 . proto is not None : <EOL> self . f1 . proto . notifyOnDisconnect ( lambda : dl [ <NUM_LIT:0> ] . callback ( None ) ) <EOL> else : <EOL> dl [ <NUM_LIT:0> ] . callback ( None ) <EOL> if self . sub is not None and self . sub . publisher is not None : <EOL> self . sub . publisher . broker . notifyOnDisconnect ( <EOL> lambda : dl [ <NUM_LIT:1> ] . callback ( None ) ) <EOL> self . sub . publisher . broker . transport . loseConnection ( ) <EOL> else : <EOL> dl [ <NUM_LIT:1> ] . callback ( None ) <EOL> if self . port1 is not None : <EOL> dl . append ( self . port1 . stopListening ( ) ) <EOL> if self . port2 is not None : <EOL> dl . append ( self . port2 . stopListening ( ) ) <EOL> return defer . gatherResults ( dl ) <EOL> def testDistrib ( self ) : <EOL> r1 = resource . Resource ( ) <EOL> r1 . putChild ( "<STR_LIT>" , static . Data ( "<STR_LIT:root>" , "<STR_LIT>" ) ) <EOL> site1 = server . Site ( r1 ) <EOL> self . f1 = PBServerFactory ( distrib . ResourcePublisher ( site1 ) ) <EOL> self . port1 = reactor . listenTCP ( <NUM_LIT:0> , self . f1 ) <EOL> self . sub = distrib . ResourceSubscription ( "<STR_LIT:127.0.0.1>" , <EOL> self . port1 . getHost ( ) . port ) <EOL> r2 = resource . Resource ( ) <EOL> r2 . putChild ( "<STR_LIT>" , self . sub ) <EOL> f2 = MySite ( r2 ) <EOL> self . port2 = reactor . listenTCP ( <NUM_LIT:0> , f2 ) <EOL> d = client . getPage ( "<STR_LIT>" % self . port2 . getHost ( ) . port ) <EOL> d . addCallback ( self . assertEqual , '<STR_LIT:root>' ) <EOL> return d <EOL> def _setupDistribServer ( self , child ) : <EOL> """<STR_LIT>""" <EOL> distribRoot = resource . Resource ( ) <EOL> distribRoot . putChild ( "<STR_LIT>" , child ) <EOL> distribSite = server . Site ( distribRoot ) <EOL> self . f1 = distribFactory = PBServerFactory ( <EOL> distrib . ResourcePublisher ( distribSite ) ) <EOL> distribPort = reactor . listenTCP ( <EOL> <NUM_LIT:0> , distribFactory , interface = "<STR_LIT:127.0.0.1>" ) <EOL> self . addCleanup ( distribPort . stopListening ) <EOL> addr = distribPort . getHost ( ) <EOL> self . sub = mainRoot = distrib . ResourceSubscription ( <EOL> addr . host , addr . port ) <EOL> mainSite = server . Site ( mainRoot ) <EOL> mainPort = reactor . listenTCP ( <NUM_LIT:0> , mainSite , interface = "<STR_LIT:127.0.0.1>" ) <EOL> self . addCleanup ( mainPort . stopListening ) <EOL> mainAddr = mainPort . getHost ( ) <EOL> return mainPort , mainAddr <EOL> def _requestTest ( self , child , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> mainPort , mainAddr = self . _setupDistribServer ( child ) <EOL> return client . getPage ( "<STR_LIT>" % ( <EOL> mainAddr . host , mainAddr . port ) , ** kwargs ) <EOL> def _requestAgentTest ( self , child , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> mainPort , mainAddr = self . _setupDistribServer ( child ) <EOL> d = client . Agent ( reactor ) . request ( "<STR_LIT:GET>" , "<STR_LIT>" % ( <EOL> mainAddr . host , mainAddr . port ) , ** kwargs ) <EOL> def cbCollectBody ( response ) : <EOL> protocol = proto_helpers . AccumulatingProtocol ( ) <EOL> response . deliverBody ( protocol ) <EOL> d = protocol . closedDeferred = defer . Deferred ( ) <EOL> d . addCallback ( lambda _ : ( protocol , response ) ) <EOL> return d <EOL> d . addCallback ( cbCollectBody ) <EOL> return d <EOL> def test_requestHeaders ( self ) : <EOL> """<STR_LIT>""" <EOL> requestHeaders = { } <EOL> class ReportRequestHeaders ( resource . Resource ) : <EOL> def render ( self , request ) : <EOL> requestHeaders . update ( dict ( <EOL> request . requestHeaders . getAllRawHeaders ( ) ) ) <EOL> return "<STR_LIT>" <EOL> request = self . _requestTest ( <EOL> ReportRequestHeaders ( ) , headers = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> def cbRequested ( result ) : <EOL> self . assertEqual ( requestHeaders [ '<STR_LIT>' ] , [ '<STR_LIT:bar>' ] ) <EOL> request . addCallback ( cbRequested ) <EOL> return request <EOL> def test_requestResponseCode ( self ) : <EOL> """<STR_LIT>""" <EOL> class SetResponseCode ( resource . Resource ) : <EOL> def render ( self , request ) : <EOL> request . setResponseCode ( <NUM_LIT:200> ) <EOL> return "<STR_LIT>" <EOL> request = self . _requestAgentTest ( SetResponseCode ( ) ) <EOL> def cbRequested ( result ) : <EOL> self . assertEqual ( result [ <NUM_LIT:0> ] . data , "<STR_LIT>" ) <EOL> self . assertEqual ( result [ <NUM_LIT:1> ] . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( result [ <NUM_LIT:1> ] . phrase , "<STR_LIT:OK>" ) <EOL> request . addCallback ( cbRequested ) <EOL> return request <EOL> def test_requestResponseCodeMessage ( self ) : <EOL> """<STR_LIT>""" <EOL> class SetResponseCode ( resource . Resource ) : <EOL> def render ( self , request ) : <EOL> request . setResponseCode ( <NUM_LIT:200> , "<STR_LIT>" ) <EOL> return "<STR_LIT>" <EOL> request = self . _requestAgentTest ( SetResponseCode ( ) ) <EOL> def cbRequested ( result ) : <EOL> self . assertEqual ( result [ <NUM_LIT:0> ] . data , "<STR_LIT>" ) <EOL> self . assertEqual ( result [ <NUM_LIT:1> ] . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( result [ <NUM_LIT:1> ] . phrase , "<STR_LIT>" ) <EOL> request . addCallback ( cbRequested ) <EOL> return request <EOL> def test_largeWrite ( self ) : <EOL> """<STR_LIT>""" <EOL> class LargeWrite ( resource . Resource ) : <EOL> def render ( self , request ) : <EOL> request . write ( '<STR_LIT:x>' * SIZE_LIMIT + '<STR_LIT:y>' ) <EOL> request . finish ( ) <EOL> return server . NOT_DONE_YET <EOL> request = self . _requestTest ( LargeWrite ( ) ) <EOL> request . addCallback ( self . assertEqual , '<STR_LIT:x>' * SIZE_LIMIT + '<STR_LIT:y>' ) <EOL> return request <EOL> def test_largeReturn ( self ) : <EOL> """<STR_LIT>""" <EOL> class LargeReturn ( resource . Resource ) : <EOL> def render ( self , request ) : <EOL> return '<STR_LIT:x>' * SIZE_LIMIT + '<STR_LIT:y>' <EOL> request = self . _requestTest ( LargeReturn ( ) ) <EOL> request . addCallback ( self . assertEqual , '<STR_LIT:x>' * SIZE_LIMIT + '<STR_LIT:y>' ) <EOL> return request <EOL> def test_connectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . f1 = serverFactory = PBServerFactory ( pb . Root ( ) ) <EOL> self . port1 = serverPort = reactor . listenTCP ( <NUM_LIT:0> , serverFactory ) <EOL> self . sub = subscription = distrib . ResourceSubscription ( <EOL> "<STR_LIT:127.0.0.1>" , serverPort . getHost ( ) . port ) <EOL> request = DummyRequest ( [ '<STR_LIT>' ] ) <EOL> d = _render ( subscription , request ) <EOL> def cbRendered ( ignored ) : <EOL> self . assertEqual ( request . responseCode , <NUM_LIT> ) <EOL> self . assertEqual ( len ( self . flushLoggedErrors ( pb . NoSuchMethod ) ) , <NUM_LIT:1> ) <EOL> d . addCallback ( cbRendered ) <EOL> return d <EOL> class _PasswordDatabase : <EOL> def __init__ ( self , users ) : <EOL> self . _users = users <EOL> def getpwall ( self ) : <EOL> return iter ( self . _users ) <EOL> def getpwnam ( self , username ) : <EOL> for user in self . _users : <EOL> if user [ <NUM_LIT:0> ] == username : <EOL> return user <EOL> raise KeyError ( ) <EOL> class UserDirectoryTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . alice = ( '<STR_LIT>' , '<STR_LIT:x>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' , self . mktemp ( ) , '<STR_LIT>' ) <EOL> self . bob = ( '<STR_LIT>' , '<STR_LIT:x>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' , self . mktemp ( ) , '<STR_LIT>' ) <EOL> self . database = _PasswordDatabase ( [ self . alice , self . bob ] ) <EOL> self . directory = distrib . UserDirectory ( self . database ) <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( verifyObject ( resource . IResource , self . directory ) ) <EOL> def _404Test ( self , name ) : <EOL> """<STR_LIT>""" <EOL> request = DummyRequest ( [ name ] ) <EOL> result = self . directory . getChild ( name , request ) <EOL> d = _render ( result , request ) <EOL> def cbRendered ( ignored ) : <EOL> self . assertEqual ( request . responseCode , <NUM_LIT> ) <EOL> d . addCallback ( cbRendered ) <EOL> return d <EOL> def test_getInvalidUser ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _404Test ( '<STR_LIT>' ) <EOL> def test_getUserWithoutResource ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _404Test ( '<STR_LIT>' ) <EOL> def test_getPublicHTMLChild ( self ) : <EOL> """<STR_LIT>""" <EOL> home = filepath . FilePath ( self . bob [ - <NUM_LIT:2> ] ) <EOL> public_html = home . child ( '<STR_LIT>' ) <EOL> public_html . makedirs ( ) <EOL> request = DummyRequest ( [ '<STR_LIT>' ] ) <EOL> result = self . directory . getChild ( '<STR_LIT>' , request ) <EOL> self . assertIsInstance ( result , static . File ) <EOL> self . assertEqual ( result . path , public_html . path ) <EOL> def test_getDistribChild ( self ) : <EOL> """<STR_LIT>""" <EOL> home = filepath . FilePath ( self . bob [ - <NUM_LIT:2> ] ) <EOL> home . makedirs ( ) <EOL> web = home . child ( '<STR_LIT>' ) <EOL> request = DummyRequest ( [ '<STR_LIT>' ] ) <EOL> result = self . directory . getChild ( '<STR_LIT>' , request ) <EOL> self . assertIsInstance ( result , distrib . ResourceSubscription ) <EOL> self . assertEqual ( result . host , '<STR_LIT>' ) <EOL> self . assertEqual ( abspath ( result . port ) , web . path ) <EOL> def test_invalidMethod ( self ) : <EOL> """<STR_LIT>""" <EOL> request = DummyRequest ( [ '<STR_LIT>' ] ) <EOL> request . method = '<STR_LIT:POST>' <EOL> self . assertRaises ( <EOL> server . UnsupportedMethod , self . directory . render , request ) <EOL> def test_render ( self ) : <EOL> """<STR_LIT>""" <EOL> public_html = filepath . FilePath ( self . alice [ - <NUM_LIT:2> ] ) . child ( '<STR_LIT>' ) <EOL> public_html . makedirs ( ) <EOL> web = filepath . FilePath ( self . bob [ - <NUM_LIT:2> ] ) <EOL> web . makedirs ( ) <EOL> web . child ( '<STR_LIT>' ) . setContent ( "<STR_LIT>" ) <EOL> request = DummyRequest ( [ '<STR_LIT>' ] ) <EOL> result = _render ( self . directory , request ) <EOL> def cbRendered ( ignored ) : <EOL> document = parseString ( '<STR_LIT>' . join ( request . written ) ) <EOL> [ alice , bob ] = document . getElementsByTagName ( '<STR_LIT>' ) <EOL> self . assertEqual ( alice . firstChild . tagName , '<STR_LIT:a>' ) <EOL> self . assertEqual ( alice . firstChild . getAttribute ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( alice . firstChild . firstChild . data , '<STR_LIT>' ) <EOL> self . assertEqual ( bob . firstChild . tagName , '<STR_LIT:a>' ) <EOL> self . assertEqual ( bob . firstChild . getAttribute ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( bob . firstChild . firstChild . data , '<STR_LIT>' ) <EOL> result . addCallback ( cbRendered ) <EOL> return result <EOL> def test_passwordDatabase ( self ) : <EOL> """<STR_LIT>""" <EOL> directory = distrib . UserDirectory ( ) <EOL> self . assertIdentical ( directory . _pwd , pwd ) <EOL> if pwd is None : <EOL> test_passwordDatabase . skip = "<STR_LIT>" </s>
