def testRoot(self):
rr = RootResource()
rr.putChild('', rr)
rr.putChild('bar', resource.Resource())
chan = self.createServer(rr)
for url in ['/', '/bar', '/bar/baz', '/bar/']:
request = server.Request(chan, 1)
request.setHost('example.com', 81)
request.gotLength(0)
request.requestReceived('GET', url, 'HTTP/1.0')
self.assertEqual(request.getRootURL(), "http://example.com/")
python类Request()的实例源码
def __init__(self, *args, **kw):
server.Request.__init__(self, *args, **kw)
self._cookieCache = {}
from cStringIO import StringIO
self.content = StringIO()
self.received_headers['host'] = 'fake.com'
self.written = StringIO()
def write(self, data):
self.written.write(data)
server.Request.write(self, data)
def addCookie(self, k, v, *args,**kw):
server.Request.addCookie(self,k,v,*args,**kw)
assert not self._cookieCache.has_key(k), "Should not be setting duplicate cookies!"
self._cookieCache[k] = v
self.channel.received_cookies[k] = v
def testChildLink(self):
request = server.Request(DummyChannel(), 1)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
self.assertEqual(request.childLink('baz'), 'bar/baz')
request = server.Request(DummyChannel(), 1)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar/', 'HTTP/1.0')
self.assertEqual(request.childLink('baz'), 'baz')
def testPrePathURLSimple(self):
request = server.Request(DummyChannel(), 1)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
request.setHost('example.com', 80)
self.assertEqual(request.prePathURL(), 'http://example.com/foo/bar')
def testPrePathURLNonDefault(self):
d = DummyChannel()
d.transport = DummyChannel.TCP()
d.transport.port = 81
request = server.Request(d, 1)
request.setHost('example.com', 81)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
self.assertEqual(request.prePathURL(), 'http://example.com:81/foo/bar')
def testPrePathURLSSLPort(self):
d = DummyChannel()
d.transport = DummyChannel.TCP()
d.transport.port = 443
request = server.Request(d, 1)
request.setHost('example.com', 443)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
self.assertEqual(request.prePathURL(), 'http://example.com:443/foo/bar')
def testPrePathURLSSLPortAndSSL(self):
d = DummyChannel()
d.transport = DummyChannel.SSL()
d.transport.port = 443
request = server.Request(d, 1)
request.setHost('example.com', 443)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
self.assertEqual(request.prePathURL(), 'https://example.com/foo/bar')
def testPrePathURLSSLNonDefault(self):
d = DummyChannel()
d.transport = DummyChannel.SSL()
d.transport.port = 81
request = server.Request(d, 1)
request.setHost('example.com', 81)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
self.assertEqual(request.prePathURL(), 'https://example.com:81/foo/bar')
def testPrePathURLSetSSLHost(self):
d = DummyChannel()
d.transport = DummyChannel.TCP()
d.transport.port = 81
request = server.Request(d, 1)
request.setHost('foo.com', 81, 1)
request.gotLength(0)
request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
self.assertEqual(request.prePathURL(), 'https://foo.com:81/foo/bar')
def testNotifyFinishConnectionLost(self):
d = DummyChannel()
d.transport = DummyChannel.TCP()
request = server.Request(d, 1)
finished = request.notifyFinish()
request.connectionLost(error.ConnectionDone("Connection done"))
return self.assertFailure(finished, error.ConnectionDone)
def testSimple(self):
r = resource.Resource()
r.isLeaf=0
rr = RootResource()
r.putChild('foo', rr)
rr.putChild('', rr)
rr.putChild('bar', resource.Resource())
chan = self.createServer(r)
for url in ['/foo/', '/foo/bar', '/foo/bar/baz', '/foo/bar/']:
request = server.Request(chan, 1)
request.setHost('example.com', 81)
request.gotLength(0)
request.requestReceived('GET', url, 'HTTP/1.0')
self.assertEqual(request.getRootURL(), "http://example.com/foo")
def testRoot(self):
rr = RootResource()
rr.putChild('', rr)
rr.putChild('bar', resource.Resource())
chan = self.createServer(rr)
for url in ['/', '/bar', '/bar/baz', '/bar/']:
request = server.Request(chan, 1)
request.setHost('example.com', 81)
request.gotLength(0)
request.requestReceived('GET', url, 'HTTP/1.0')
self.assertEqual(request.getRootURL(), "http://example.com/")
def __init__(self, *args, **kw):
server.Request.__init__(self, *args, **kw)
self._cookieCache = {}
from cStringIO import StringIO
self.content = StringIO()
self.received_headers['host'] = 'fake.com'
self.written = StringIO()
def write(self, data):
self.written.write(data)
server.Request.write(self, data)
def addCookie(self, k, v, *args,**kw):
server.Request.addCookie(self,k,v,*args,**kw)
assert not self._cookieCache.has_key(k), "Should not be setting duplicate cookies!"
self._cookieCache[k] = v
self.channel.received_cookies[k] = v
def testNoUICaching(self):
ui_caching = False
site = CacheControlledSite(ui_caching, self._resource)
request = server.Request(DummyChannel(), False)
request.prepath = [b""]
request.postpath = [b""]
site.getResourceFor(request)
self.assertTrue(request.responseHeaders.getRawHeaders("cache-control") == ["no-store, must-revalidate"])
def testUICaching(self):
ui_caching = True
site = CacheControlledSite(ui_caching, self._resource)
request = server.Request(DummyChannel(), False)
request.prepath = [b""]
request.postpath = [b""]
site.getResourceFor(request)
self.assertTrue(request.responseHeaders.getRawHeaders("cache-control") == ["max-age={}".format(FRESHNESS_TIME_SECS)])
def __init__(self, *args, **kw):
server.Request.__init__(self, *args, **kw)
self._logger = logging.getLogger(self.__class__.__name__)
