def test_responder_resource_child(self):
"""
When a GET request is made to the ACME challenge path, and the
responder resource has a child resource at the correct path, the value
of the resource should be returned.
"""
self.responder_resource.putChild(b'foo', Data(b'bar', 'text/plain'))
response = self.client.get(
'http://localhost/.well-known/acme-challenge/foo')
assert_that(response, succeeded(MatchesAll(
MatchesStructure(
code=Equals(200),
headers=HasHeader('Content-Type', ['text/plain'])),
After(methodcaller('content'), succeeded(Equals(b'bar')))
)))
# Sanity check that a request to a different subpath does not succeed
response = self.client.get(
'http://localhost/.well-known/acme-challenge/baz')
assert_that(response, succeeded(MatchesStructure(code=Equals(404))))
python类Data()的实例源码
def testDistrib(self):
# site1 is the publisher
r1 = resource.Resource()
r1.putChild("there", static.Data("root", "text/plain"))
site1 = server.Site(r1)
self.f1 = PBServerFactory(distrib.ResourcePublisher(site1))
self.port1 = reactor.listenTCP(0, self.f1)
self.sub = distrib.ResourceSubscription("127.0.0.1",
self.port1.getHost().port)
r2 = resource.Resource()
r2.putChild("here", self.sub)
f2 = MySite(r2)
self.port2 = reactor.listenTCP(0, f2)
d = client.getPage("http://127.0.0.1:%d/here/there" % \
self.port2.getHost().port)
d.addCallback(self.failUnlessEqual, 'root')
return d
def setUp(self):
plainRoot = static.Data('not me', 'text/plain')
tlsRoot = static.Data('me neither', 'text/plain')
plainSite = server.Site(plainRoot, timeout=None)
tlsSite = server.Site(tlsRoot, timeout=None)
from twisted import test
self.tlsPort = reactor.listenSSL(0, tlsSite,
contextFactory=ssl.DefaultOpenSSLContextFactory(
sibpath(test.__file__, 'server.pem'),
sibpath(test.__file__, 'server.pem'),
),
interface="127.0.0.1")
self.plainPort = reactor.listenTCP(0, plainSite, interface="127.0.0.1")
self.plainPortno = self.plainPort.getHost().port
self.tlsPortno = self.tlsPort.getHost().port
plainRoot.putChild('one', util.Redirect(self.getHTTPS('two')))
tlsRoot.putChild('two', util.Redirect(self.getHTTP('three')))
plainRoot.putChild('three', util.Redirect(self.getHTTPS('four')))
tlsRoot.putChild('four', static.Data('FOUND IT!', 'text/plain'))
def testDistrib(self):
# site1 is the publisher
r1 = resource.Resource()
r1.putChild("there", static.Data("root", "text/plain"))
site1 = server.Site(r1)
self.f1 = PBServerFactory(distrib.ResourcePublisher(site1))
self.port1 = reactor.listenTCP(0, self.f1)
self.sub = distrib.ResourceSubscription("127.0.0.1",
self.port1.getHost().port)
r2 = resource.Resource()
r2.putChild("here", self.sub)
f2 = MySite(r2)
self.port2 = reactor.listenTCP(0, f2)
d = client.getPage("http://127.0.0.1:%d/here/there" % \
self.port2.getHost().port)
d.addCallback(self.failUnlessEqual, 'root')
return d
def setUp(self):
plainRoot = static.Data('not me', 'text/plain')
tlsRoot = static.Data('me neither', 'text/plain')
plainSite = server.Site(plainRoot, timeout=None)
tlsSite = server.Site(tlsRoot, timeout=None)
from twisted import test
self.tlsPort = reactor.listenSSL(0, tlsSite,
contextFactory=ssl.DefaultOpenSSLContextFactory(
sibpath(test.__file__, 'server.pem'),
sibpath(test.__file__, 'server.pem'),
),
interface="127.0.0.1")
self.plainPort = reactor.listenTCP(0, plainSite, interface="127.0.0.1")
self.plainPortno = self.plainPort.getHost().port
self.tlsPortno = self.tlsPort.getHost().port
plainRoot.putChild('one', util.Redirect(self.getHTTPS('two')))
tlsRoot.putChild('two', util.Redirect(self.getHTTP('three')))
plainRoot.putChild('three', util.Redirect(self.getHTTPS('four')))
tlsRoot.putChild('four', static.Data('FOUND IT!', 'text/plain'))
def test_renderWithHost(self):
"""
L{NameVirtualHost.render} returns the result of rendering the resource
which is the value in the instance's C{host} dictionary corresponding
to the key indicated by the value of the I{Host} header in the request.
"""
virtualHostResource = NameVirtualHost()
virtualHostResource.addHost(b'example.org', Data(b"winner", ""))
request = DummyRequest([b''])
request.requestHeaders.addRawHeader(b'host', b'example.org')
d = _render(virtualHostResource, request)
def cbRendered(ignored, request):
self.assertEqual(b''.join(request.written), b"winner")
d.addCallback(cbRendered, request)
# The port portion of the Host header should not be considered.
requestWithPort = DummyRequest([b''])
requestWithPort.requestHeaders.addRawHeader(b'host', b'example.org:8000')
dWithPort = _render(virtualHostResource, requestWithPort)
def cbRendered(ignored, requestWithPort):
self.assertEqual(b''.join(requestWithPort.written), b"winner")
dWithPort.addCallback(cbRendered, requestWithPort)
return gatherResults([d, dWithPort])
def setUp(self):
"""
Create a realm, portal, and L{HTTPAuthSessionWrapper} to use in the tests.
"""
self.username = b'foo bar'
self.password = b'bar baz'
self.avatarContent = b"contents of the avatar resource itself"
self.childName = b"foo-child"
self.childContent = b"contents of the foo child of the avatar"
self.checker = InMemoryUsernamePasswordDatabaseDontUse()
self.checker.addUser(self.username, self.password)
self.avatar = Data(self.avatarContent, 'text/plain')
self.avatar.putChild(
self.childName, Data(self.childContent, 'text/plain'))
self.avatars = {self.username: self.avatar}
self.realm = Realm(self.avatars.get)
self.portal = portal.Portal(self.realm, [self.checker])
self.credentialFactories = []
self.wrapper = HTTPAuthSessionWrapper(
self.portal, self.credentialFactories)
def testDistrib(self):
# site1 is the publisher
r1 = resource.Resource()
r1.putChild("there", static.Data("root", "text/plain"))
site1 = server.Site(r1)
self.f1 = PBServerFactory(distrib.ResourcePublisher(site1))
self.port1 = reactor.listenTCP(0, self.f1)
self.sub = distrib.ResourceSubscription("127.0.0.1",
self.port1.getHost().port)
r2 = resource.Resource()
r2.putChild("here", self.sub)
f2 = MySite(r2)
self.port2 = reactor.listenTCP(0, f2)
agent = client.Agent(reactor)
d = agent.request(b"GET", "http://127.0.0.1:%d/here/there" % \
(self.port2.getHost().port,))
d.addCallback(client.readBody)
d.addCallback(self.assertEqual, 'root')
return d
def setUp(self):
plainRoot = Data(b'not me', 'text/plain')
tlsRoot = Data(b'me neither', 'text/plain')
plainSite = server.Site(plainRoot, timeout=None)
tlsSite = server.Site(tlsRoot, timeout=None)
self.tlsPort = reactor.listenSSL(
0, tlsSite,
contextFactory=ssl.DefaultOpenSSLContextFactory(
serverPEMPath, serverPEMPath),
interface="127.0.0.1")
self.plainPort = reactor.listenTCP(0, plainSite, interface="127.0.0.1")
self.plainPortno = self.plainPort.getHost().port
self.tlsPortno = self.tlsPort.getHost().port
plainRoot.putChild(b'one', Redirect(self.getHTTPS('two')))
tlsRoot.putChild(b'two', Redirect(self.getHTTP('three')))
plainRoot.putChild(b'three', Redirect(self.getHTTPS('four')))
tlsRoot.putChild(b'four', Data(b'FOUND IT!', 'text/plain'))
def test_getPageDeprecated(self):
"""
L{client.getPage} is deprecated.
"""
port = reactor.listenTCP(
0, server.Site(Data(b'', 'text/plain')), interface="127.0.0.1")
portno = port.getHost().port
self.addCleanup(port.stopListening)
url = networkString("http://127.0.0.1:%d" % (portno,))
d = client.getPage(url)
warningInfo = self.flushWarnings([self.test_getPageDeprecated])
self.assertEqual(len(warningInfo), 1)
self.assertEqual(warningInfo[0]['category'], DeprecationWarning)
self.assertEqual(
warningInfo[0]['message'],
"twisted.web.client.getPage was deprecated in "
"Twisted 16.7.0; please use https://pypi.org/project/treq/ or twisted.web.client.Agent instead")
return d.addErrback(lambda _: None)
def testStaticPaths(self):
import os
dp = os.path.join(self.mktemp(),"hello")
ddp = os.path.join(dp, "goodbye")
tp = os.path.abspath(os.path.join(dp,"world.txt"))
tpy = os.path.join(dp,"wyrld.rpy")
os.makedirs(dp)
f = open(tp,"wb")
f.write("hello world")
f = open(tpy, "wb")
f.write("""
from twisted.web.static import Data
resource = Data('dynamic world','text/plain')
""")
f = static.File(dp)
f.processors = {
'.rpy': script.ResourceScript,
}
f.indexNames = f.indexNames + ['world.txt']
self.assertEquals(f.getChild('', DummyRequest([''])).path,
tp)
self.assertEquals(f.getChild('wyrld.rpy', DummyRequest(['wyrld.rpy'])
).__class__,
static.Data)
f = static.File(dp)
wtextr = DummyRequest(['world.txt'])
wtext = f.getChild('world.txt', wtextr)
self.assertEquals(wtext.path, tp)
wtext.render(wtextr)
self.assertEquals(wtextr.outgoingHeaders.get('content-length'),
str(len('hello world')))
self.assertNotEquals(f.getChild('', DummyRequest([''])).__class__,
static.File)
def setUp(self):
root = static.Data('El toro!', 'text/plain')
root.putChild("cookiemirror", CookieMirrorResource())
root.putChild("rawcookiemirror", RawCookieMirrorResource())
site = server.Site(root, timeout=None)
self.port = self._listen(site)
self.portno = self.port.getHost().port
def testSessionInit(self):
sessWrapped = static.Data("you should never see this", "text/plain")
swChild = static.Data("NO", "text/plain")
sessWrapped.putChild("yyy",swChild)
swrap = guard.SessionWrapper(sessWrapped)
da = static.Data("b","text/plain")
da.putChild("xxx", swrap)
st = FakeSite(da)
chan = FakeHTTPChannel()
chan.site = st
# first we're going to make sure that the session doesn't get set by
# accident when browsing without first explicitly initializing the
# session
req = FakeHTTPRequest(chan, queued=0)
req.requestReceived("GET", "/xxx/yyy", "1.0")
assert len(req._cookieCache.values()) == 0, req._cookieCache.values()
self.assertEquals(req.getSession(),None)
# now we're going to make sure that the redirect and cookie are properly set
req = FakeHTTPRequest(chan, queued=0)
req.requestReceived("GET", "/xxx/"+guard.INIT_SESSION, "1.0")
ccv = req._cookieCache.values()
self.assertEquals(len(ccv),1)
cookie = ccv[0]
# redirect set?
self.failUnless(req.headers.has_key('location'))
# redirect matches cookie?
self.assertEquals(req.headers['location'].split('/')[-1], guard.SESSION_KEY+cookie)
# URL is correct?
self.assertEquals(req.headers['location'],
'http://fake.com/xxx/'+guard.SESSION_KEY+cookie)
oldreq = req
# now let's try with a request for the session-cookie URL that has a cookie set
url = "/"+(oldreq.headers['location'].split('http://fake.com/',1))[1]
req = chan.makeFakeRequest(url)
self.assertEquals(req.headers['location'].split('?')[0],
'http://fake.com/xxx/')
for sz in swrap.sessions.values():
sz.expire()
def setUp(self):
market_factory = WebSocketServerFactory('ws://localhost:8080/market_stream')
user_factory = WebSocketServerFactory('ws://localhost:8080/user_stream')
market_factory.protocol = MarketStreamServerProtocol
user_factory.protocol = UserStreamServerProtocol
market_factory.startFactory()
user_factory.startFactory()
root = Data('', 'text/plain')
root.putChild(b'market_stream', WebSocketResource(market_factory))
root.putChild(b'user_stream', WebSocketResource(user_factory))
site = Site(root)
reactor.listenTCP(8080, site)
def run_server():
reactor.run(installSignalHandlers=False)
Thread(target=run_server).start()
def testStaticPaths(self):
import os
dp = os.path.join(self.mktemp(),"hello")
ddp = os.path.join(dp, "goodbye")
tp = os.path.abspath(os.path.join(dp,"world.txt"))
tpy = os.path.join(dp,"wyrld.rpy")
os.makedirs(dp)
f = open(tp,"wb")
f.write("hello world")
f = open(tpy, "wb")
f.write("""
from twisted.web.static import Data
resource = Data('dynamic world','text/plain')
""")
f = static.File(dp)
f.processors = {
'.rpy': script.ResourceScript,
}
f.indexNames = f.indexNames + ['world.txt']
self.assertEquals(f.getChild('', DummyRequest([''])).path,
tp)
self.assertEquals(f.getChild('wyrld.rpy', DummyRequest(['wyrld.rpy'])
).__class__,
static.Data)
f = static.File(dp)
wtextr = DummyRequest(['world.txt'])
wtext = f.getChild('world.txt', wtextr)
self.assertEquals(wtext.path, tp)
wtext.render(wtextr)
self.assertEquals(wtextr.outgoingHeaders.get('content-length'),
str(len('hello world')))
self.assertNotEquals(f.getChild('', DummyRequest([''])).__class__,
static.File)
def setUp(self):
root = static.Data('El toro!', 'text/plain')
root.putChild("cookiemirror", CookieMirrorResource())
root.putChild("rawcookiemirror", RawCookieMirrorResource())
site = server.Site(root, timeout=None)
self.port = self._listen(site)
self.portno = self.port.getHost().port
def testSessionInit(self):
sessWrapped = static.Data("you should never see this", "text/plain")
swChild = static.Data("NO", "text/plain")
sessWrapped.putChild("yyy",swChild)
swrap = guard.SessionWrapper(sessWrapped)
da = static.Data("b","text/plain")
da.putChild("xxx", swrap)
st = FakeSite(da)
chan = FakeHTTPChannel()
chan.site = st
# first we're going to make sure that the session doesn't get set by
# accident when browsing without first explicitly initializing the
# session
req = FakeHTTPRequest(chan, queued=0)
req.requestReceived("GET", "/xxx/yyy", "1.0")
assert len(req._cookieCache.values()) == 0, req._cookieCache.values()
self.assertEquals(req.getSession(),None)
# now we're going to make sure that the redirect and cookie are properly set
req = FakeHTTPRequest(chan, queued=0)
req.requestReceived("GET", "/xxx/"+guard.INIT_SESSION, "1.0")
ccv = req._cookieCache.values()
self.assertEquals(len(ccv),1)
cookie = ccv[0]
# redirect set?
self.failUnless(req.headers.has_key('location'))
# redirect matches cookie?
self.assertEquals(req.headers['location'].split('/')[-1], guard.SESSION_KEY+cookie)
# URL is correct?
self.assertEquals(req.headers['location'],
'http://fake.com/xxx/'+guard.SESSION_KEY+cookie)
oldreq = req
# now let's try with a request for the session-cookie URL that has a cookie set
url = "/"+(oldreq.headers['location'].split('http://fake.com/',1))[1]
req = chan.makeFakeRequest(url)
self.assertEquals(req.headers['location'].split('?')[0],
'http://fake.com/xxx/')
for sz in swrap.sessions.values():
sz.expire()
def test_getChild(self):
"""
L{_HostResource.getChild} returns the proper I{Resource} for the vhost
embedded in the URL. Verify that returning the proper I{Resource}
required changing the I{Host} in the header.
"""
bazroot = Data(b'root data', "")
bazuri = Data(b'uri data', "")
baztest = Data(b'test data', "")
bazuri.putChild(b'test', baztest)
bazroot.putChild(b'uri', bazuri)
hr = _HostResource()
root = NameVirtualHost()
root.default = Data(b'default data', "")
root.addHost(b'baz.com', bazroot)
request = DummyRequest([b'uri', b'test'])
request.prepath = [b'bar', b'http', b'baz.com']
request.site = Site(root)
request.isSecure = lambda: False
request.host = b''
step = hr.getChild(b'baz.com', request) # Consumes rest of path
self.assertIsInstance(step, Data)
request = DummyRequest([b'uri', b'test'])
step = root.getChild(b'uri', request)
self.assertIsInstance(step, NoResource)
def test_renderWithoutHost(self):
"""
L{NameVirtualHost.render} returns the result of rendering the
instance's C{default} if it is not L{None} and there is no I{Host}
header in the request.
"""
virtualHostResource = NameVirtualHost()
virtualHostResource.default = Data(b"correct result", "")
request = DummyRequest([''])
self.assertEqual(
virtualHostResource.render(request), b"correct result")
def test_renderWithUnknownHost(self):
"""
L{NameVirtualHost.render} returns the result of rendering the
instance's C{default} if it is not L{None} and there is no host
matching the value of the I{Host} header in the request.
"""
virtualHostResource = NameVirtualHost()
virtualHostResource.default = Data(b"correct data", "")
request = DummyRequest([b''])
request.requestHeaders.addRawHeader(b'host', b'example.com')
d = _render(virtualHostResource, request)
def cbRendered(ignored):
self.assertEqual(b''.join(request.written), b"correct data")
d.addCallback(cbRendered)
return d
def test_getChild(self):
"""
L{NameVirtualHost.getChild} returns correct I{Resource} based off
the header and modifies I{Request} to ensure proper prepath and
postpath are set.
"""
virtualHostResource = NameVirtualHost()
leafResource = Data(b"leaf data", "")
leafResource.isLeaf = True
normResource = Data(b"norm data", "")
virtualHostResource.addHost(b'leaf.example.org', leafResource)
virtualHostResource.addHost(b'norm.example.org', normResource)
request = DummyRequest([])
request.requestHeaders.addRawHeader(b'host', b'norm.example.org')
request.prepath = [b'']
self.assertIsInstance(virtualHostResource.getChild(b'', request),
NoResource)
self.assertEqual(request.prepath, [b''])
self.assertEqual(request.postpath, [])
request = DummyRequest([])
request.requestHeaders.addRawHeader(b'host', b'leaf.example.org')
request.prepath = [b'']
self.assertIsInstance(virtualHostResource.getChild(b'', request),
Data)
self.assertEqual(request.prepath, [])
self.assertEqual(request.postpath, [b''])
def test_invalidMethod(self):
"""
L{Data.render} raises L{UnsupportedMethod} in response to a non-I{GET},
non-I{HEAD} request.
"""
data = static.Data(b"foo", b"bar")
request = DummyRequest([b''])
request.method = b'POST'
self.assertRaises(UnsupportedMethod, data.render, request)
def test_processors(self):
"""
If a request is made which encounters a L{File} before a final segment
which names a file with an extension which is in the L{File}'s
C{processors} mapping, the processor associated with that extension is
used to serve the response to the request.
"""
base = FilePath(self.mktemp())
base.makedirs()
base.child("foo.bar").setContent(
b"from twisted.web.static import Data\n"
b"resource = Data(b'dynamic world', 'text/plain')\n")
file = static.File(base.path)
file.processors = {b'.bar': script.ResourceScript}
request = DummyRequest([b"foo.bar"])
child = resource.getChildForRequest(file, request)
d = self._render(child, request)
def cbRendered(ignored):
self.assertEqual(b''.join(request.written), b'dynamic world')
self.assertEqual(
request.responseHeaders.getRawHeaders(b'content-length')[0],
b'13')
d.addCallback(cbRendered)
return d
def setUp(self):
self.channel = DummyChannel()
staticResource = Data(b"Some data", "text/plain")
wrapped = resource.EncodingResourceWrapper(
staticResource, [server.GzipEncoderFactory()])
self.channel.site.resource.putChild(b"foo", wrapped)
def setUp(self):
self.resource = static.Data(
b"This text is not a valid XML-RPC response.",
b"text/plain")
self.resource.isLeaf = True
self.port = reactor.listenTCP(0, server.Site(self.resource),
interface='127.0.0.1')
def setUp(self):
self.agent = None # for twisted.web.client.Agent test
self.cleanupServerConnections = 0
r = resource.Resource()
r.putChild(b"file", Data(b"0123456789", "text/html"))
r.putChild(b"redirect", Redirect(b"/file"))
self.infiniteRedirectResource = CountingRedirect(b"/infiniteRedirect")
r.putChild(b"infiniteRedirect", self.infiniteRedirectResource)
r.putChild(b"wait", ForeverTakingResource())
r.putChild(b"write-then-wait", ForeverTakingResource(write=True))
r.putChild(b"never-read", ForeverTakingNoReadingResource())
r.putChild(b"error", ErrorResource())
r.putChild(b"nolength", NoLengthResource())
r.putChild(b"host", HostHeaderResource())
r.putChild(b"payload", PayloadResource())
r.putChild(b"broken", BrokenDownloadResource())
r.putChild(b"cookiemirror", CookieMirrorResource())
r.putChild(b'delay1', DelayResource(1))
r.putChild(b'delay2', DelayResource(2))
self.afterFoundGetCounter = CountingResource()
r.putChild(b"afterFoundGetCounter", self.afterFoundGetCounter)
r.putChild(b"afterFoundGetRedirect", Redirect(b"/afterFoundGetCounter"))
miscasedHead = Data(b"miscased-head GET response content", "major/minor")
miscasedHead.render_Head = lambda request: b"miscased-head content"
r.putChild(b"miscased-head", miscasedHead)
self.extendedRedirect = ExtendedRedirect(b'/extendedRedirect')
r.putChild(b"extendedRedirect", self.extendedRedirect)
self.site = server.Site(r, timeout=None)
self.wrapper = WrappingFactory(self.site)
self.port = self._listen(self.wrapper)
self.portno = self.port.getHost().port
def setUp(self):
root = Data(b'El toro!', 'text/plain')
root.putChild(b"cookiemirror", CookieMirrorResource())
root.putChild(b"rawcookiemirror", RawCookieMirrorResource())
site = server.Site(root, timeout=None)
self.port = self._listen(site)
self.portno = self.port.getHost().port
def test_rpyProcessor(self):
"""
The I{--path} option creates a root resource which serves the
C{resource} global defined by the Python source in any child with
the C{".rpy"} extension.
"""
path, root = self._pathOption()
path.child("foo.rpy").setContent(
b"from twisted.web.static import Data\n"
b"resource = Data('content', 'major/minor')\n")
child = root.getChild("foo.rpy", None)
self.assertIsInstance(child, Data)
self.assertEqual(child.data, 'content')
self.assertEqual(child.type, 'major/minor')
def test_makePersonalServerFactory(self):
"""
L{makePersonalServerFactory} returns a PB server factory which has
as its root object a L{ResourcePublisher}.
"""
# The fact that this pile of objects can actually be used somehow is
# verified by twisted.web.test.test_distrib.
site = Site(Data(b"foo bar", "text/plain"))
serverFactory = makePersonalServerFactory(site)
self.assertIsInstance(serverFactory, PBServerFactory)
self.assertIsInstance(serverFactory.root, ResourcePublisher)
self.assertIdentical(serverFactory.root.site, site)
def __init__(self):
static.Data.__init__(
self,
b"""
<html>
<head><title>Twisted Web Demo</title><head>
<body>
Hello! This is a Twisted Web test page.
</body>
</html>
""",
"text/html")
