我们从Python开源项目中，提取了以下32个代码示例，用于说明如何使用twisted.web.static.Data()。
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))))
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 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")
def start_fake_riak_server(self, stats): def cb(listener): self.addCleanup(listener.stopListening) return listener data = static.Data(json.dumps(stats).encode(), 'application/json') data.isLeaf = True site = server.Site(data) endpoint = endpoints.TCP4ServerEndpoint(reactor, 0) return endpoint.listen(site).addCallback(cb)
def check_tls_config(self, ca_key, ca_cert, get_kubernetes): """ Verify that a TLS server configured with the given key and certificate and the Kubernetes client returned by ``get_kubernetes`` can negotiate a TLS connection. """ # Set up an HTTPS server that requires the certificate chain from the # configuration file. This, because there's no way to pry inside a # Context and inspect its state nor any easy way to make Agent talk # over an in-memory transport. from twisted.internet import reactor endpoint = SSL4ServerEndpoint( reactor, 0, CertificateOptions( privateKey=ca_key.original, certificate=ca_cert.original, trustRoot=trustRootFromCertificates([ca_cert]), ), ) root = Resource() root.putChild(b"", Data(b"success", b"text/plain")) # Construct the Kubernetes client objects with a Redirectable reactor. # This is necessary because the URL we pass to the Agent we get needs # to agree with the configuration file that was already written (or it # won't select the right client certificate). Just one of the many # reasons it would be better if we didn't have to do real networking # here. redirectable = Redirectable(reactor) client = get_kubernetes(redirectable).client() agent = client.agent d = endpoint.listen(Site(root)) def listening(port): self.addCleanup(port.stopListening) redirectable.set_redirect(port.getHost().host, port.getHost().port) url = b"https://127.0.0.1:8443/" return agent.request(b"GET", url) d.addCallback(listening) return d
