FM / test

/opt/imh-python/lib/python3.9/site-packages/tornado/test/

/opt/imh-python/lib/python3.9/site-packages/tornado/test

Name	Size	Mode	Actions
csv_translations/	-	0755	rm
gettext_translations/	-	0755	rm
static/	-	0755	rm
templates/	-	0755	rm
__pycache__/	-	0755	rm
asyncio_test.py	7329	0644	edit dl rm
auth_test.py	24026	0644	edit dl rm
autoreload_test.py	4075	0644	edit dl rm
concurrent_test.py	6174	0644	edit dl rm
curl_httpclient_test.py	4477	0644	edit dl rm
escape_test.py	12694	0644	edit dl rm
gen_test.py	34721	0644	edit dl rm
http1connection_test.py	1972	0644	edit dl rm
httpclient_test.py	32067	0644	edit dl rm
httpserver_test.py	46332	0644	edit dl rm
httputil_test.py	19600	0644	edit dl rm
import_test.py	2046	0644	edit dl rm
ioloop_test.py	26060	0644	edit dl rm
iostream_test.py	46178	0644	edit dl rm
locale_test.py	5907	0644	edit dl rm
locks_test.py	17564	0644	edit dl rm
log_test.py	9778	0644	edit dl rm
netutil_test.py	8047	0644	edit dl rm
options_test.cfg	76	0644	edit dl rm
options_test.py	12174	0644	edit dl rm
options_test_types.cfg	277	0644	edit dl rm
options_test_types_str.cfg	158	0644	edit dl rm
process_test.py	11392	0644	edit dl rm
queues_test.py	14188	0644	edit dl rm
resolve_test_helper.py	421	0644	edit dl rm
routing_test.py	9121	0644	edit dl rm
runtests.py	8440	0644	edit dl rm
simple_httpclient_test.py	31821	0644	edit dl rm
static_foo.txt	97	0644	edit dl rm
tcpclient_test.py	17115	0644	edit dl rm
tcpserver_test.py	6674	0644	edit dl rm
template_test.py	19204	0644	edit dl rm
test.crt	1244	0644	edit dl rm
test.key	1732	0644	edit dl rm
testing_test.py	10985	0644	edit dl rm
twisted_test.py	8116	0644	edit dl rm
util.py	3768	0644	edit dl rm
util_test.py	10127	0644	edit dl rm
websocket_test.py	28331	0644	edit dl rm
web_test.py	118423	0644	edit dl rm
windows_test.py	698	0644	edit dl rm
wsgi_test.py	677	0644	edit dl rm
__init__.py	398	0644	edit dl rm
__main__.py	347	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/tornado/test/twisted_test.py (8116B)

# Author: Ovidiu Predescu
# Date: July 2011
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import asyncio
import logging
import signal
import unittest
import warnings

from tornado.escape import utf8
from tornado import gen
from tornado.httpclient import AsyncHTTPClient
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.testing import bind_unused_port, AsyncTestCase, gen_test
from tornado.web import RequestHandler, Application

try:
    from twisted.internet.defer import (  # type: ignore
        Deferred,
        inlineCallbacks,
        returnValue,
    )
    from twisted.internet.protocol import Protocol  # type: ignore
    from twisted.internet.asyncioreactor import AsyncioSelectorReactor  # type: ignore
    from twisted.web.client import Agent, readBody  # type: ignore
    from twisted.web.resource import Resource  # type: ignore
    from twisted.web.server import Site  # type: ignore

have_twisted = True
except ImportError:
    have_twisted = False
else:
    # Not used directly but needed for `yield deferred` to work.
    import tornado.platform.twisted  # noqa: F401

skipIfNoTwisted = unittest.skipUnless(have_twisted, "twisted module not present")

def save_signal_handlers():
    saved = {}
    for sig in [signal.SIGINT, signal.SIGTERM, signal.SIGCHLD]:
        saved[sig] = signal.getsignal(sig)
    if "twisted" in repr(saved):
        # This indicates we're not cleaning up after ourselves properly.
        raise Exception("twisted signal handlers already installed")
    return saved

def restore_signal_handlers(saved):
    for sig, handler in saved.items():
        signal.signal(sig, handler)

# Test various combinations of twisted and tornado http servers,
# http clients, and event loop interfaces.

@skipIfNoTwisted
class CompatibilityTests(unittest.TestCase):
    def setUp(self):
        self.saved_signals = save_signal_handlers()
        self.io_loop = IOLoop()
        self.io_loop.make_current()
        self.reactor = AsyncioSelectorReactor()

def tearDown(self):
        self.reactor.disconnectAll()
        self.io_loop.clear_current()
        self.io_loop.close(all_fds=True)
        restore_signal_handlers(self.saved_signals)

def start_twisted_server(self):
        class HelloResource(Resource):
            isLeaf = True

def render_GET(self, request):
                return b"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_tornado_server(self):
        class HelloHandler(RequestHandler):
            def get(self):
                self.write("Hello from tornado!")

app = Application([("/", HelloHandler)], log_function=lambda x: None)
        server = HTTPServer(app)
        sock, self.tornado_port = bind_unused_port()
        server.add_sockets([sock])

def run_reactor(self):
        # In theory, we can run the event loop through Tornado,
        # Twisted, or asyncio interfaces. However, since we're trying
        # to avoid installing anything as the global event loop, only
        # the twisted interface gets everything wired up correectly
        # without extra hacks. This method is a part of a
        # no-longer-used generalization that allowed us to test
        # different combinations.
        self.stop_loop = self.reactor.stop
        self.stop = self.reactor.stop
        self.reactor.run()

def tornado_fetch(self, url, runner):
        client = AsyncHTTPClient()
        fut = asyncio.ensure_future(client.fetch(url))
        fut.add_done_callback(lambda f: self.stop_loop())
        runner()
        return fut.result()

def twisted_fetch(self, url, runner):
        # http://twistedmatrix.com/documents/current/web/howto/client.html
        chunks = []
        client = Agent(self.reactor)
        d = client.request(b"GET", utf8(url))

class Accumulator(Protocol):
            def __init__(self, finished):
                self.finished = finished

def dataReceived(self, data):
                chunks.append(data)

def connectionLost(self, reason):
                self.finished.callback(None)

def callback(response):
            finished = Deferred()
            response.deliverBody(Accumulator(finished))
            return finished

d.addCallback(callback)

def shutdown(failure):
            if hasattr(self, "stop_loop"):
                self.stop_loop()
            elif failure is not None:
                # loop hasn't been initialized yet; try our best to
                # get an error message out. (the runner() interaction
                # should probably be refactored).
                try:
                    failure.raiseException()
                except:
                    logging.error("exception before starting loop", exc_info=True)

d.addBoth(shutdown)
        runner()
        self.assertTrue(chunks)
        return b"".join(chunks)

def twisted_coroutine_fetch(self, url, runner):
        body = [None]

@gen.coroutine
        def f():
            # This is simpler than the non-coroutine version, but it cheats
            # by reading the body in one blob instead of streaming it with
            # a Protocol.
            client = Agent(self.reactor)
            response = yield client.request(b"GET", utf8(url))
            with warnings.catch_warnings():
                # readBody has a buggy DeprecationWarning in Twisted 15.0:
                # https://twistedmatrix.com/trac/changeset/43379
                warnings.simplefilter("ignore", category=DeprecationWarning)
                body[0] = yield readBody(response)
            self.stop_loop()

self.io_loop.add_callback(f)
        runner()
        return body[0]

def testTwistedServerTornadoClientReactor(self):
        self.start_twisted_server()
        response = self.tornado_fetch(
            "http://127.0.0.1:%d" % self.twisted_port, self.run_reactor
        )
        self.assertEqual(response.body, b"Hello from twisted!")

def testTornadoServerTwistedClientReactor(self):
        self.start_tornado_server()
        response = self.twisted_fetch(
            "http://127.0.0.1:%d" % self.tornado_port, self.run_reactor
        )
        self.assertEqual(response, b"Hello from tornado!")

def testTornadoServerTwistedCoroutineClientReactor(self):
        self.start_tornado_server()
        response = self.twisted_coroutine_fetch(
            "http://127.0.0.1:%d" % self.tornado_port, self.run_reactor
        )
        self.assertEqual(response, b"Hello from tornado!")

@skipIfNoTwisted
class ConvertDeferredTest(AsyncTestCase):
    @gen_test
    def test_success(self):
        @inlineCallbacks
        def fn():
            if False:
                # inlineCallbacks doesn't work with regular functions;
                # must have a yield even if it's unreachable.
                yield
            returnValue(42)

res = yield fn()
        self.assertEqual(res, 42)

@gen_test
    def test_failure(self):
        @inlineCallbacks
        def fn():
            if False:
                yield
            1 / 0

with self.assertRaises(ZeroDivisionError):
            yield fn()

if __name__ == "__main__":
    unittest.main()

cmd: