Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/urllib3/util/connection.py : 32%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import absolute_import
3import socket
5from urllib3.exceptions import LocationParseError
7from ..contrib import _appengine_environ
8from ..packages import six
9from .wait import NoWayToWaitForSocketError, wait_for_read
12def is_connection_dropped(conn): # Platform-specific
13 """
14 Returns True if the connection is dropped and should be closed.
16 :param conn:
17 :class:`http.client.HTTPConnection` object.
19 Note: For platforms like AppEngine, this will always return ``False`` to
20 let the platform handle connection recycling transparently for us.
21 """
22 sock = getattr(conn, "sock", False)
23 if sock is False: # Platform-specific: AppEngine
24 return False
25 if sock is None: # Connection already closed (such as by httplib).
26 return True
27 try:
28 # Returns True if readable, which here means it's been dropped
29 return wait_for_read(sock, timeout=0.0)
30 except NoWayToWaitForSocketError: # Platform-specific: AppEngine
31 return False
34# This function is copied from socket.py in the Python 2.7 standard
35# library test suite. Added to its signature is only `socket_options`.
36# One additional modification is that we avoid binding to IPv6 servers
37# discovered in DNS if the system doesn't have IPv6 functionality.
38def create_connection(
39 address,
40 timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
41 source_address=None,
42 socket_options=None,
43):
44 """Connect to *address* and return the socket object.
46 Convenience function. Connect to *address* (a 2-tuple ``(host,
47 port)``) and return the socket object. Passing the optional
48 *timeout* parameter will set the timeout on the socket instance
49 before attempting to connect. If no *timeout* is supplied, the
50 global default timeout setting returned by :func:`socket.getdefaulttimeout`
51 is used. If *source_address* is set it must be a tuple of (host, port)
52 for the socket to bind as a source address before making the connection.
53 An host of '' or port 0 tells the OS to use the default.
54 """
56 host, port = address
57 if host.startswith("["):
58 host = host.strip("[]")
59 err = None
61 # Using the value from allowed_gai_family() in the context of getaddrinfo lets
62 # us select whether to work with IPv4 DNS records, IPv6 records, or both.
63 # The original create_connection function always returns all records.
64 family = allowed_gai_family()
66 try:
67 host.encode("idna")
68 except UnicodeError:
69 return six.raise_from(
70 LocationParseError(u"'%s', label empty or too long" % host), None
71 )
73 for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
74 af, socktype, proto, canonname, sa = res
75 sock = None
76 try:
77 sock = socket.socket(af, socktype, proto)
79 # If provided, set socket level options before connecting.
80 _set_socket_options(sock, socket_options)
82 if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
83 sock.settimeout(timeout)
84 if source_address:
85 sock.bind(source_address)
86 sock.connect(sa)
87 return sock
89 except socket.error as e:
90 err = e
91 if sock is not None:
92 sock.close()
93 sock = None
95 if err is not None:
96 raise err
98 raise socket.error("getaddrinfo returns an empty list")
101def _set_socket_options(sock, options):
102 if options is None:
103 return
105 for opt in options:
106 sock.setsockopt(*opt)
109def allowed_gai_family():
110 """This function is designed to work in the context of
111 getaddrinfo, where family=socket.AF_UNSPEC is the default and
112 will perform a DNS search for both IPv6 and IPv4 records."""
114 family = socket.AF_INET
115 if HAS_IPV6:
116 family = socket.AF_UNSPEC
117 return family
120def _has_ipv6(host):
121 """ Returns True if the system can bind an IPv6 address. """
122 sock = None
123 has_ipv6 = False
125 # App Engine doesn't support IPV6 sockets and actually has a quota on the
126 # number of sockets that can be used, so just early out here instead of
127 # creating a socket needlessly.
128 # See https://github.com/urllib3/urllib3/issues/1446
129 if _appengine_environ.is_appengine_sandbox():
130 return False
132 if socket.has_ipv6:
133 # has_ipv6 returns true if cPython was compiled with IPv6 support.
134 # It does not tell us if the system has IPv6 support enabled. To
135 # determine that we must bind to an IPv6 address.
136 # https://github.com/urllib3/urllib3/pull/611
137 # https://bugs.python.org/issue658327
138 try:
139 sock = socket.socket(socket.AF_INET6)
140 sock.bind((host, 0))
141 has_ipv6 = True
142 except Exception:
143 pass
145 if sock:
146 sock.close()
147 return has_ipv6
150HAS_IPV6 = _has_ipv6("::1")