.local/lib/python2.7/site-packages/trollius/test_support.py

   1 # Subset of test.support from CPython 3.5, just what we need to run asyncio
   2 # test suite. The code is copied from CPython 3.5 to not depend on the test
   3 # module because it is rarely installed.
   4
   5 # Ignore symbol TEST_HOME_DIR: test_events works without it
   6
   7 from __future__ import absolute_import
   8 import functools
   9 import gc
  10 import os
  11 import platform
  12 import re
  13 import socket
  14 import subprocess
  15 import sys
  16 import time
  17
  18 from trollius import test_utils
  19
  20 # A constant likely larger than the underlying OS pipe buffer size, to
  21 # make writes blocking.
  22 # Windows limit seems to be around 512 B, and many Unix kernels have a
  23 # 64 KiB pipe buffer size or 16 * PAGE_SIZE: take a few megs to be sure.
  24 # (see issue #17835 for a discussion of this number).
  25 PIPE_MAX_SIZE = 4 * 1024 * 1024 + 1
  26
  27 def strip_python_stderr(stderr):
  28     """Strip the stderr of a Python process from potential debug output
  29     emitted by the interpreter.
  30
  31     This will typically be run on the result of the communicate() method
  32     of a subprocess.Popen object.
  33     """
  34     stderr = re.sub(br"\[\d+ refs, \d+ blocks\]\r?\n?", b"", stderr).strip()
  35     return stderr
  36
  37
  38 # Executing the interpreter in a subprocess
  39 def _assert_python(expected_success, *args, **env_vars):
  40     if '__isolated' in env_vars:
  41         isolated = env_vars.pop('__isolated')
  42     else:
  43         isolated = not env_vars
  44     cmd_line = [sys.executable]
  45     if sys.version_info >= (3, 3):
  46         cmd_line.extend(('-X', 'faulthandler'))
  47     if isolated and sys.version_info >= (3, 4):
  48         # isolated mode: ignore Python environment variables, ignore user
  49         # site-packages, and don't add the current directory to sys.path
  50         cmd_line.append('-I')
  51     elif not env_vars:
  52         # ignore Python environment variables
  53         cmd_line.append('-E')
  54     # Need to preserve the original environment, for in-place testing of
  55     # shared library builds.
  56     env = os.environ.copy()
  57     # But a special flag that can be set to override -- in this case, the
  58     # caller is responsible to pass the full environment.
  59     if env_vars.pop('__cleanenv', None):
  60         env = {}
  61     env.update(env_vars)
  62     cmd_line.extend(args)
  63     p = subprocess.Popen(cmd_line, stdin=subprocess.PIPE,
  64                          stdout=subprocess.PIPE, stderr=subprocess.PIPE,
  65                          env=env)
  66     try:
  67         out, err = p.communicate()
  68     finally:
  69         subprocess._cleanup()
  70         p.stdout.close()
  71         p.stderr.close()
  72     rc = p.returncode
  73     err = strip_python_stderr(err)
  74     if (rc and expected_success) or (not rc and not expected_success):
  75         raise AssertionError(
  76             "Process return code is %d, "
  77             "stderr follows:\n%s" % (rc, err.decode('ascii', 'ignore')))
  78     return rc, out, err
  79
  80
  81 def assert_python_ok(*args, **env_vars):
  82     """
  83     Assert that running the interpreter with `args` and optional environment
  84     variables `env_vars` succeeds (rc == 0) and return a (return code, stdout,
  85     stderr) tuple.
  86
  87     If the __cleanenv keyword is set, env_vars is used a fresh environment.
  88
  89     Python is started in isolated mode (command line option -I),
  90     except if the __isolated keyword is set to False.
  91     """
  92     return _assert_python(True, *args, **env_vars)
  93
  94
  95 is_jython = sys.platform.startswith('java')
  96
  97 def gc_collect():
  98     """Force as many objects as possible to be collected.
  99
 100     In non-CPython implementations of Python, this is needed because timely
 101     deallocation is not guaranteed by the garbage collector.  (Even in CPython
 102     this can be the case in case of reference cycles.)  This means that __del__
 103     methods may be called later than expected and weakrefs may remain alive for
 104     longer than expected.  This function tries its best to force all garbage
 105     objects to disappear.
 106     """
 107     gc.collect()
 108     if is_jython:
 109         time.sleep(0.1)
 110     gc.collect()
 111     gc.collect()
 112
 113
 114 HOST = "127.0.0.1"
 115 HOSTv6 = "::1"
 116
 117
 118 def _is_ipv6_enabled():
 119     """Check whether IPv6 is enabled on this host."""
 120     if socket.has_ipv6:
 121         sock = None
 122         try:
 123             sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
 124             sock.bind((HOSTv6, 0))
 125             return True
 126         except OSError:
 127             pass
 128         finally:
 129             if sock:
 130                 sock.close()
 131     return False
 132
 133 IPV6_ENABLED = _is_ipv6_enabled()
 134
 135
 136 def find_unused_port(family=socket.AF_INET, socktype=socket.SOCK_STREAM):
 137     """Returns an unused port that should be suitable for binding.  This is
 138     achieved by creating a temporary socket with the same family and type as
 139     the 'sock' parameter (default is AF_INET, SOCK_STREAM), and binding it to
 140     the specified host address (defaults to 0.0.0.0) with the port set to 0,
 141     eliciting an unused ephemeral port from the OS.  The temporary socket is
 142     then closed and deleted, and the ephemeral port is returned.
 143
 144     Either this method or bind_port() should be used for any tests where a
 145     server socket needs to be bound to a particular port for the duration of
 146     the test.  Which one to use depends on whether the calling code is creating
 147     a python socket, or if an unused port needs to be provided in a constructor
 148     or passed to an external program (i.e. the -accept argument to openssl's
 149     s_server mode).  Always prefer bind_port() over find_unused_port() where
 150     possible.  Hard coded ports should *NEVER* be used.  As soon as a server
 151     socket is bound to a hard coded port, the ability to run multiple instances
 152     of the test simultaneously on the same host is compromised, which makes the
 153     test a ticking time bomb in a buildbot environment. On Unix buildbots, this
 154     may simply manifest as a failed test, which can be recovered from without
 155     intervention in most cases, but on Windows, the entire python process can
 156     completely and utterly wedge, requiring someone to log in to the buildbot
 157     and manually kill the affected process.
 158
 159     (This is easy to reproduce on Windows, unfortunately, and can be traced to
 160     the SO_REUSEADDR socket option having different semantics on Windows versus
 161     Unix/Linux.  On Unix, you can't have two AF_INET SOCK_STREAM sockets bind,
 162     listen and then accept connections on identical host/ports.  An EADDRINUSE
 163     OSError will be raised at some point (depending on the platform and
 164     the order bind and listen were called on each socket).
 165
 166     However, on Windows, if SO_REUSEADDR is set on the sockets, no EADDRINUSE
 167     will ever be raised when attempting to bind two identical host/ports. When
 168     accept() is called on each socket, the second caller's process will steal
 169     the port from the first caller, leaving them both in an awkwardly wedged
 170     state where they'll no longer respond to any signals or graceful kills, and
 171     must be forcibly killed via OpenProcess()/TerminateProcess().
 172
 173     The solution on Windows is to use the SO_EXCLUSIVEADDRUSE socket option
 174     instead of SO_REUSEADDR, which effectively affords the same semantics as
 175     SO_REUSEADDR on Unix.  Given the propensity of Unix developers in the Open
 176     Source world compared to Windows ones, this is a common mistake.  A quick
 177     look over OpenSSL's 0.9.8g source shows that they use SO_REUSEADDR when
 178     openssl.exe is called with the 's_server' option, for example. See
 179     http://bugs.python.org/issue2550 for more info.  The following site also
 180     has a very thorough description about the implications of both REUSEADDR
 181     and EXCLUSIVEADDRUSE on Windows:
 182     http://msdn2.microsoft.com/en-us/library/ms740621(VS.85).aspx)
 183
 184     XXX: although this approach is a vast improvement on previous attempts to
 185     elicit unused ports, it rests heavily on the assumption that the ephemeral
 186     port returned to us by the OS won't immediately be dished back out to some
 187     other process when we close and delete our temporary socket but before our
 188     calling code has a chance to bind the returned port.  We can deal with this
 189     issue if/when we come across it.
 190     """
 191
 192     tempsock = socket.socket(family, socktype)
 193     port = bind_port(tempsock)
 194     tempsock.close()
 195     del tempsock
 196     return port
 197
 198 def bind_port(sock, host=HOST):
 199     """Bind the socket to a free port and return the port number.  Relies on
 200     ephemeral ports in order to ensure we are using an unbound port.  This is
 201     important as many tests may be running simultaneously, especially in a
 202     buildbot environment.  This method raises an exception if the sock.family
 203     is AF_INET and sock.type is SOCK_STREAM, *and* the socket has SO_REUSEADDR
 204     or SO_REUSEPORT set on it.  Tests should *never* set these socket options
 205     for TCP/IP sockets.  The only case for setting these options is testing
 206     multicasting via multiple UDP sockets.
 207
 208     Additionally, if the SO_EXCLUSIVEADDRUSE socket option is available (i.e.
 209     on Windows), it will be set on the socket.  This will prevent anyone else
 210     from bind()'ing to our host/port for the duration of the test.
 211     """
 212
 213     if sock.family == socket.AF_INET and sock.type == socket.SOCK_STREAM:
 214         if hasattr(socket, 'SO_REUSEADDR'):
 215             if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 1:
 216                 raise TestFailed("tests should never set the SO_REUSEADDR "
 217                                  "socket option on TCP/IP sockets!")
 218         if hasattr(socket, 'SO_REUSEPORT'):
 219             try:
 220                 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT)
 221                 if reuse == 1:
 222                     raise TestFailed("tests should never set the SO_REUSEPORT "
 223                                      "socket option on TCP/IP sockets!")
 224             except OSError:
 225                 # Python's socket module was compiled using modern headers
 226                 # thus defining SO_REUSEPORT but this process is running
 227                 # under an older kernel that does not support SO_REUSEPORT.
 228                 pass
 229         if hasattr(socket, 'SO_EXCLUSIVEADDRUSE'):
 230             sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
 231
 232     sock.bind((host, 0))
 233     port = sock.getsockname()[1]
 234     return port
 235
 236 def requires_mac_ver(*min_version):
 237     """Decorator raising SkipTest if the OS is Mac OS X and the OS X
 238     version if less than min_version.
 239
 240     For example, @requires_mac_ver(10, 5) raises SkipTest if the OS X version
 241     is lesser than 10.5.
 242     """
 243     def decorator(func):
 244         @functools.wraps(func)
 245         def wrapper(*args, **kw):
 246             if sys.platform == 'darwin':
 247                 version_txt = platform.mac_ver()[0]
 248                 try:
 249                     version = tuple(map(int, version_txt.split('.')))
 250                 except ValueError:
 251                     pass
 252                 else:
 253                     if version < min_version:
 254                         min_version_txt = '.'.join(map(str, min_version))
 255                         raise test_utils.SkipTest(
 256                             "Mac OS X %s or higher required, not %s"
 257                             % (min_version_txt, version_txt))
 258             return func(*args, **kw)
 259         wrapper.min_version = min_version
 260         return wrapper
 261     return decorator
 262
 263 def _requires_unix_version(sysname, min_version):
 264     """Decorator raising SkipTest if the OS is `sysname` and the version is
 265     less than `min_version`.
 266
 267     For example, @_requires_unix_version('FreeBSD', (7, 2)) raises SkipTest if
 268     the FreeBSD version is less than 7.2.
 269     """
 270     def decorator(func):
 271         @functools.wraps(func)
 272         def wrapper(*args, **kw):
 273             if platform.system() == sysname:
 274                 version_txt = platform.release().split('-', 1)[0]
 275                 try:
 276                     version = tuple(map(int, version_txt.split('.')))
 277                 except ValueError:
 278                     pass
 279                 else:
 280                     if version < min_version:
 281                         min_version_txt = '.'.join(map(str, min_version))
 282                         raise test_utils.SkipTest(
 283                             "%s version %s or higher required, not %s"
 284                             % (sysname, min_version_txt, version_txt))
 285             return func(*args, **kw)
 286         wrapper.min_version = min_version
 287         return wrapper
 288     return decorator
 289
 290 def requires_freebsd_version(*min_version):
 291     """Decorator raising SkipTest if the OS is FreeBSD and the FreeBSD version
 292     is less than `min_version`.
 293
 294     For example, @requires_freebsd_version(7, 2) raises SkipTest if the FreeBSD
 295     version is less than 7.2.
 296     """
 297     return _requires_unix_version('FreeBSD', min_version)
 298
 299 # Use test.support if available
 300 try:
 301     from test.support import *
 302 except ImportError:
 303     pass
 304
 305 # Use test.script_helper if available
 306 try:
 307     from test.script_helper import assert_python_ok
 308 except ImportError:
 309     pass