Package ldaptor :: Package test :: Module mockweb
[hide private]
[frames] | no frames]

Source Code for Module ldaptor.test.mockweb

 1  """I mock the web.""" 
 2   
 3  from nevow import url 
 4  from twisted.internet import address 
 5  from twisted.web import client 
 6   
 7  from nevow import testutil 
 8  from ldaptor.test import util 
 9   
10 -class FakeChannel(testutil.FakeChannel):
11 - def requestDone(self, request):
12 self.transport.loseConnection()
13
14 -class MyHTTPPageGetter(client.HTTPPageGetter):
15 - def handleStatus_301(self):
16 if not self.followRedirect: 17 client.HTTPPageGetter.handleStatus_301(self) 18 return 19 20 l = self.headers.get('location') 21 if not l: 22 self.handleStatusDefault() 23 url = l[0] 24 self.factory.setURL(url) 25 26 _getPage_connect(clientFactory=self.factory, 27 serverAddress=address.IPv4Address( 28 'TCP', self.factory.host, self.factory.port), 29 clientAddress=None) 30 self.quietLoss = 1 31 self.transport.loseConnection()
32
33 -class HTTPClientFactory_noCookies(client.HTTPClientFactory):
34 - def gotHeaders(self, headers):
35 client.HTTPClientFactory.gotHeaders(self, headers) 36 self.cookies.clear()
37
38 -def _getPage_connect(clientFactory, 39 serverAddress, 40 clientAddress):
41 clientProto = clientFactory.buildProtocol(serverAddress) 42 serverProto = clientFactory.site.buildProtocol(clientAddress) 43 pump = util.returnConnected(serverProto, clientProto, 44 serverAddress=serverAddress, 45 clientAddress=clientAddress)
46
47 -def getPage(site, u, extraInfo=False, 48 factoryClass=client.HTTPClientFactory, 49 *a, **kw):
50 u = url.URL.fromString(str(u)) 51 clientFactory = factoryClass(str(u), *a, **kw) 52 clientFactory.protocol = MyHTTPPageGetter 53 clientFactory.site = site 54 if ':' in u.netloc: 55 host, port = u.netloc.split(':', 1) 56 else: 57 host, port = u.netloc, 80 58 serverAddress = address.IPv4Address('TCP', host, port) 59 clientAddress = address.IPv4Address('TCP', 'localhost', 1024) 60 _getPage_connect(clientFactory, serverAddress, clientAddress) 61 62 if extraInfo: 63 def _cb(page): 64 return {'status': clientFactory.status, 65 'version': clientFactory.version, 66 'message': clientFactory.message, 67 'headers': clientFactory.headers, 68 'page': page, 69 'url': url.URL.fromString(clientFactory.url), 70 }
71 clientFactory.deferred.addCallback(_cb) 72 return clientFactory.deferred 73
74 -def getPage_noCookies(*a, **kw):
75 defaults = { 76 'factoryClass': HTTPClientFactory_noCookies, 77 } 78 defaults.update(kw) 79 return getPage(*a, **defaults)
80