File: test_resolvers.py

package info (click to toggle)
python-tornado 6.5.2-1
  • links: PTS, VCS
  • area: main
  • in suites: experimental
  • size: 3,176 kB
  • sloc: python: 28,920; javascript: 156; sh: 100; ansic: 72; xml: 49; makefile: 49; sql: 23
file content (64 lines) | stat: -rwxr-xr-x 1,725 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
#!/usr/bin/env python
"""Basic test for Tornado resolvers.

Queries real domain names and prints the results from each resolver.
Requires a working internet connection, which is why it's not in a
unit test.

Will be removed in Tornado 7.0 when the pluggable resolver system is
removed.
"""
import pprint
import socket

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.netutil import Resolver, ThreadedResolver, DefaultExecutorResolver
from tornado.options import parse_command_line, define, options

try:
    import pycares
except ImportError:
    pycares = None

define(
    "family", default="unspec", help="Address family to query: unspec, inet, or inet6"
)


@gen.coroutine
def main():
    args = parse_command_line()

    if not args:
        args = ["localhost", "www.google.com", "www.facebook.com", "www.dropbox.com"]

    resolvers = [Resolver(), ThreadedResolver(), DefaultExecutorResolver()]

    if pycares is not None:
        from tornado.platform.caresresolver import CaresResolver

        resolvers.append(CaresResolver())

    family = {
        "unspec": socket.AF_UNSPEC,
        "inet": socket.AF_INET,
        "inet6": socket.AF_INET6,
    }[options.family]

    for host in args:
        print("Resolving %s" % host)
        for resolver in resolvers:
            try:
                addrinfo = yield resolver.resolve(host, 80, family)
            except Exception as e:
                print("%s: %s: %s" % (resolver.__class__.__name__, type(e), e))
            else:
                print(
                    "%s: %s" % (resolver.__class__.__name__, pprint.pformat(addrinfo))
                )
        print()


if __name__ == "__main__":
    IOLoop.instance().run_sync(main)