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)
python类Site()的实例源码
def test_downloadPageDeprecated(self):
"""
L{client.downloadPage} 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,))
path = FilePath(self.mktemp())
d = client.downloadPage(url, path.path)
warningInfo = self.flushWarnings([self.test_downloadPageDeprecated])
self.assertEqual(len(warningInfo), 1)
self.assertEqual(warningInfo[0]['category'], DeprecationWarning)
self.assertEqual(
warningInfo[0]['message'],
"twisted.web.client.downloadPage 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 main():
parser = argparse.ArgumentParser()
parser.add_argument('resource')
args = parser.parse_args()
module_name, name = args.resource.rsplit('.', 1)
sys.path.append('.')
resource = getattr(import_module(module_name), name)()
http_port = reactor.listenTCP(PORT, Site(resource))
def print_listening():
host = http_port.getHost()
print('Mock server {} running at http://{}:{}'.format(
resource, host.host, host.port))
reactor.callWhenRunning(print_listening)
reactor.run()
def start(web_cfg):
"""
Start the web service
"""
static = Static(web_cfg)
root_resource = Root(web_cfg, static)
site = Site(root_resource)
return site
def run(self, handler):
from twisted.web import server, wsgi
from twisted.python.threadpool import ThreadPool
from twisted.internet import reactor
thread_pool = ThreadPool()
thread_pool.start()
reactor.addSystemEventTrigger('after', 'shutdown', thread_pool.stop)
factory = server.Site(wsgi.WSGIResource(reactor, thread_pool, handler))
reactor.listenTCP(self.port, factory, interface=self.host)
if not reactor.running:
reactor.run()
def listen(self, reactor, endpoint_description):
"""
Run the server, i.e. start listening for requests on the given host and
port.
:param reactor: The ``IReactorTCP`` to use.
:param endpoint_description:
The Twisted description for the endpoint to listen on.
:return:
A deferred that returns an object that provides ``IListeningPort``.
"""
endpoint = serverFromString(reactor, endpoint_description)
return endpoint.listen(Site(self.app.resource()))
def start_twisted_server(self):
class HelloResource(Resource):
isLeaf = True
def render_GET(self, request):
return "Hello from twisted!"
site = Site(HelloResource())
port = self.reactor.listenTCP(0, site, interface='127.0.0.1')
self.twisted_port = port.getHost().port
def start_twisted_server(self):
class HelloResource(Resource):
isLeaf = True
def render_GET(self, request):
return "Hello from twisted!"
site = Site(HelloResource())
port = self.reactor.listenTCP(0, site, interface='127.0.0.1')
self.twisted_port = port.getHost().port
def start_twisted_server(self):
class HelloResource(Resource):
isLeaf = True
def render_GET(self, request):
return "Hello from twisted!"
site = Site(HelloResource())
port = self.reactor.listenTCP(0, site, interface='127.0.0.1')
self.twisted_port = port.getHost().port
def setUp(self):
BaseTestCase.setUp(self)
self.gw = twisted.TwistedGateway(expose_request=False)
root = resource.Resource()
root.putChild('', self.gw)
self.p = reactor.listenTCP(0, server.Site(root), interface="127.0.0.1")
self.port = self.p.getHost().port
def testSimplestSite(self):
sres1 = SimpleResource()
sres2 = SimpleResource()
sres1.putChild("",sres2)
site = server.Site(sres1)
assert site.getResourceFor(DummyRequest([''])) is sres2, "Got the wrong resource."
def setUp(self):
self.site = server.Site(SimpleResource())
def createServer(self, r):
chan = DummyChannel()
chan.transport = DummyChannel.TCP()
chan.site = server.Site(r)
return chan
def setUp(self):
self.p = reactor.listenTCP(0, server.Site(Test()),
interface="127.0.0.1")
self.port = self.p.getHost().port
def setUp(self):
self.p = reactor.listenTCP(
0, server.Site(Test(allowNone=True)), interface="127.0.0.1")
self.port = self.p.getHost().port
def setUp(self):
self.p = reactor.listenTCP(0, server.Site(TestAuthHeader()),
interface="127.0.0.1")
self.port = self.p.getHost().port
def setUp(self):
xmlrpc = Test()
addIntrospection(xmlrpc)
self.p = reactor.listenTCP(0, server.Site(xmlrpc),interface="127.0.0.1")
self.port = self.p.getHost().port
def setUp(self):
self.resource = static.File(__file__)
self.resource.isLeaf = True
self.port = reactor.listenTCP(0, server.Site(self.resource), interface='127.0.0.1')
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 getResourceFor(self, req):
res = server.Site.getResourceFor(self,req)
self.caughtRes = res
return res
