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/netutil_test.py (8047B)

import errno
import os
import signal
import socket
from subprocess import Popen
import sys
import time
import unittest

from tornado.netutil import (
    BlockingResolver,
    OverrideResolver,
    ThreadedResolver,
    is_valid_ip,
    bind_sockets,
)
from tornado.testing import AsyncTestCase, gen_test, bind_unused_port
from tornado.test.util import skipIfNoNetwork

import typing

if typing.TYPE_CHECKING:
    from typing import List  # noqa: F401

try:
    import pycares  # type: ignore
except ImportError:
    pycares = None
else:
    from tornado.platform.caresresolver import CaresResolver

try:
    import twisted  # type: ignore
    import twisted.names  # type: ignore
except ImportError:
    twisted = None
else:
    from tornado.platform.twisted import TwistedResolver

class _ResolverTestMixin(object):
    @gen_test
    def test_localhost(self):
        addrinfo = yield self.resolver.resolve("localhost", 80, socket.AF_UNSPEC)
        self.assertIn((socket.AF_INET, ("127.0.0.1", 80)), addrinfo)

# It is impossible to quickly and consistently generate an error in name
# resolution, so test this case separately, using mocks as needed.
class _ResolverErrorTestMixin(object):
    @gen_test
    def test_bad_host(self):
        with self.assertRaises(IOError):
            yield self.resolver.resolve("an invalid domain", 80, socket.AF_UNSPEC)

def _failing_getaddrinfo(*args):
    """Dummy implementation of getaddrinfo for use in mocks"""
    raise socket.gaierror(errno.EIO, "mock: lookup failed")

@skipIfNoNetwork
class BlockingResolverTest(AsyncTestCase, _ResolverTestMixin):
    def setUp(self):
        super(BlockingResolverTest, self).setUp()
        self.resolver = BlockingResolver()

# getaddrinfo-based tests need mocking to reliably generate errors;
# some configurations are slow to produce errors and take longer than
# our default timeout.
class BlockingResolverErrorTest(AsyncTestCase, _ResolverErrorTestMixin):
    def setUp(self):
        super(BlockingResolverErrorTest, self).setUp()
        self.resolver = BlockingResolver()
        self.real_getaddrinfo = socket.getaddrinfo
        socket.getaddrinfo = _failing_getaddrinfo

def tearDown(self):
        socket.getaddrinfo = self.real_getaddrinfo
        super(BlockingResolverErrorTest, self).tearDown()

class OverrideResolverTest(AsyncTestCase, _ResolverTestMixin):
    def setUp(self):
        super(OverrideResolverTest, self).setUp()
        mapping = {
            ("google.com", 80): ("1.2.3.4", 80),
            ("google.com", 80, socket.AF_INET): ("1.2.3.4", 80),
            ("google.com", 80, socket.AF_INET6): (
                "2a02:6b8:7c:40c:c51e:495f:e23a:3",
                80,
            ),
        }
        self.resolver = OverrideResolver(BlockingResolver(), mapping)

@gen_test
    def test_resolve_multiaddr(self):
        result = yield self.resolver.resolve("google.com", 80, socket.AF_INET)
        self.assertIn((socket.AF_INET, ("1.2.3.4", 80)), result)

result = yield self.resolver.resolve("google.com", 80, socket.AF_INET6)
        self.assertIn(
            (socket.AF_INET6, ("2a02:6b8:7c:40c:c51e:495f:e23a:3", 80, 0, 0)), result
        )

@skipIfNoNetwork
class ThreadedResolverTest(AsyncTestCase, _ResolverTestMixin):
    def setUp(self):
        super(ThreadedResolverTest, self).setUp()
        self.resolver = ThreadedResolver()

def tearDown(self):
        self.resolver.close()
        super(ThreadedResolverTest, self).tearDown()

class ThreadedResolverErrorTest(AsyncTestCase, _ResolverErrorTestMixin):
    def setUp(self):
        super(ThreadedResolverErrorTest, self).setUp()
        self.resolver = BlockingResolver()
        self.real_getaddrinfo = socket.getaddrinfo
        socket.getaddrinfo = _failing_getaddrinfo

def tearDown(self):
        socket.getaddrinfo = self.real_getaddrinfo
        super(ThreadedResolverErrorTest, self).tearDown()

@skipIfNoNetwork
@unittest.skipIf(sys.platform == "win32", "preexec_fn not available on win32")
class ThreadedResolverImportTest(unittest.TestCase):
    def test_import(self):
        TIMEOUT = 5

# Test for a deadlock when importing a module that runs the
        # ThreadedResolver at import-time. See resolve_test.py for
        # full explanation.
        command = [sys.executable, "-c", "import tornado.test.resolve_test_helper"]

start = time.time()
        popen = Popen(command, preexec_fn=lambda: signal.alarm(TIMEOUT))
        while time.time() - start < TIMEOUT:
            return_code = popen.poll()
            if return_code is not None:
                self.assertEqual(0, return_code)
                return  # Success.
            time.sleep(0.05)

self.fail("import timed out")

# We do not test errors with CaresResolver:
# Some DNS-hijacking ISPs (e.g. Time Warner) return non-empty results
# with an NXDOMAIN status code.  Most resolvers treat this as an error;
# C-ares returns the results, making the "bad_host" tests unreliable.
# C-ares will try to resolve even malformed names, such as the
# name with spaces used in this test.
@skipIfNoNetwork
@unittest.skipIf(pycares is None, "pycares module not present")
class CaresResolverTest(AsyncTestCase, _ResolverTestMixin):
    def setUp(self):
        super(CaresResolverTest, self).setUp()
        self.resolver = CaresResolver()

# TwistedResolver produces consistent errors in our test cases so we
# could test the regular and error cases in the same class. However,
# in the error cases it appears that cleanup of socket objects is
# handled asynchronously and occasionally results in "unclosed socket"
# warnings if not given time to shut down (and there is no way to
# explicitly shut it down). This makes the test flaky, so we do not
# test error cases here.
@skipIfNoNetwork
@unittest.skipIf(twisted is None, "twisted module not present")
@unittest.skipIf(
    getattr(twisted, "__version__", "0.0") < "12.1", "old version of twisted"
)
class TwistedResolverTest(AsyncTestCase, _ResolverTestMixin):
    def setUp(self):
        super(TwistedResolverTest, self).setUp()
        self.resolver = TwistedResolver()

class IsValidIPTest(unittest.TestCase):
    def test_is_valid_ip(self):
        self.assertTrue(is_valid_ip("127.0.0.1"))
        self.assertTrue(is_valid_ip("4.4.4.4"))
        self.assertTrue(is_valid_ip("::1"))
        self.assertTrue(is_valid_ip("2620:0:1cfe:face:b00c::3"))
        self.assertTrue(not is_valid_ip("www.google.com"))
        self.assertTrue(not is_valid_ip("localhost"))
        self.assertTrue(not is_valid_ip("4.4.4.4<"))
        self.assertTrue(not is_valid_ip(" 127.0.0.1"))
        self.assertTrue(not is_valid_ip(""))
        self.assertTrue(not is_valid_ip(" "))
        self.assertTrue(not is_valid_ip("\n"))
        self.assertTrue(not is_valid_ip("\x00"))

class TestPortAllocation(unittest.TestCase):
    def test_same_port_allocation(self):
        if "TRAVIS" in os.environ:
            self.skipTest("dual-stack servers often have port conflicts on travis")
        sockets = bind_sockets(0, "localhost")
        try:
            port = sockets[0].getsockname()[1]
            self.assertTrue(all(s.getsockname()[1] == port for s in sockets[1:]))
        finally:
            for sock in sockets:
                sock.close()

@unittest.skipIf(
        not hasattr(socket, "SO_REUSEPORT"), "SO_REUSEPORT is not supported"
    )
    def test_reuse_port(self):
        sockets = []  # type: List[socket.socket]
        socket, port = bind_unused_port(reuse_port=True)
        try:
            sockets = bind_sockets(port, "127.0.0.1", reuse_port=True)
            self.assertTrue(all(s.getsockname()[1] == port for s in sockets))
        finally:
            socket.close()
            for sock in sockets:
                sock.close()

cmd: