642 lines
18 KiB
Python
642 lines
18 KiB
Python
|
from __future__ import print_function, absolute_import, division
|
||
|
|
import re
|
||
|
|
import sys
|
||
|
|
import os
|
||
|
|
import traceback
|
||
|
|
import unittest
|
||
|
|
import threading
|
||
|
|
import subprocess
|
||
|
|
from time import sleep
|
||
|
|
|
||
|
|
from . import six
|
||
|
|
from gevent._config import validate_bool
|
||
|
|
from gevent._compat import perf_counter
|
||
|
|
from gevent.monkey import get_original
|
||
|
|
|
||
|
|
# pylint: disable=broad-except,attribute-defined-outside-init
|
||
|
|
|
||
|
|
BUFFER_OUTPUT = False
|
||
|
|
# This is set by the testrunner, defaulting to true (be quiet)
|
||
|
|
# But if we're run standalone, default to false
|
||
|
|
QUIET = validate_bool(os.environ.get('GEVENTTEST_QUIET', '0'))
|
||
|
|
|
||
|
|
|
||
|
|
class Popen(subprocess.Popen):
|
||
|
|
"""
|
||
|
|
Depending on when we're imported and if the process has been monkey-patched,
|
||
|
|
this could use cooperative or native Popen.
|
||
|
|
"""
|
||
|
|
timer = None # a threading.Timer instance
|
||
|
|
|
||
|
|
def __enter__(self):
|
||
|
|
return self
|
||
|
|
|
||
|
|
def __exit__(self, *args):
|
||
|
|
kill(self)
|
||
|
|
|
||
|
|
|
||
|
|
# Coloring code based on zope.testrunner
|
||
|
|
|
||
|
|
# These colors are carefully chosen to have enough contrast
|
||
|
|
# on terminals with both black and white background.
|
||
|
|
_colorscheme = {
|
||
|
|
'normal': 'normal',
|
||
|
|
'default': 'default',
|
||
|
|
|
||
|
|
'actual-output': 'red',
|
||
|
|
'character-diffs': 'magenta',
|
||
|
|
'debug': 'cyan',
|
||
|
|
'diff-chunk': 'magenta',
|
||
|
|
'error': 'brightred',
|
||
|
|
'error-number': 'brightred',
|
||
|
|
'exception': 'red',
|
||
|
|
'expected-output': 'green',
|
||
|
|
'failed-example': 'cyan',
|
||
|
|
'filename': 'lightblue',
|
||
|
|
'info': 'normal',
|
||
|
|
'lineno': 'lightred',
|
||
|
|
'number': 'green',
|
||
|
|
'ok-number': 'green',
|
||
|
|
'skipped': 'brightyellow',
|
||
|
|
'slow-test': 'brightmagenta',
|
||
|
|
'suboptimal-behaviour': 'magenta',
|
||
|
|
'testname': 'lightcyan',
|
||
|
|
'warning': 'cyan',
|
||
|
|
}
|
||
|
|
|
||
|
|
_prefixes = [
|
||
|
|
('dark', '0;'),
|
||
|
|
('light', '1;'),
|
||
|
|
('bright', '1;'),
|
||
|
|
('bold', '1;'),
|
||
|
|
]
|
||
|
|
|
||
|
|
_colorcodes = {
|
||
|
|
'default': 0,
|
||
|
|
'normal': 0,
|
||
|
|
'black': 30,
|
||
|
|
'red': 31,
|
||
|
|
'green': 32,
|
||
|
|
'brown': 33, 'yellow': 33,
|
||
|
|
'blue': 34,
|
||
|
|
'magenta': 35,
|
||
|
|
'cyan': 36,
|
||
|
|
'grey': 37, 'gray': 37, 'white': 37
|
||
|
|
}
|
||
|
|
|
||
|
|
def _color_code(color):
|
||
|
|
prefix_code = ''
|
||
|
|
for prefix, code in _prefixes:
|
||
|
|
if color.startswith(prefix):
|
||
|
|
color = color[len(prefix):]
|
||
|
|
prefix_code = code
|
||
|
|
break
|
||
|
|
color_code = _colorcodes[color]
|
||
|
|
return '\033[%s%sm' % (prefix_code, color_code)
|
||
|
|
|
||
|
|
def _color(what):
|
||
|
|
return _color_code(_colorscheme[what])
|
||
|
|
|
||
|
|
def _colorize(what, message, normal='normal'):
|
||
|
|
return _color(what) + message + _color(normal)
|
||
|
|
|
||
|
|
def log(message, *args, **kwargs):
|
||
|
|
"""
|
||
|
|
Log a *message*
|
||
|
|
|
||
|
|
:keyword str color: One of the values from _colorscheme
|
||
|
|
"""
|
||
|
|
color = kwargs.pop('color', 'normal')
|
||
|
|
|
||
|
|
if args:
|
||
|
|
string = message % args
|
||
|
|
else:
|
||
|
|
string = message
|
||
|
|
string = _colorize(color, string)
|
||
|
|
|
||
|
|
with output_lock: # pylint:disable=not-context-manager
|
||
|
|
sys.stderr.write(string + '\n')
|
||
|
|
|
||
|
|
def debug(message, *args, **kwargs):
|
||
|
|
"""
|
||
|
|
Log the *message* only if we're not in quiet mode.
|
||
|
|
"""
|
||
|
|
if not QUIET:
|
||
|
|
kwargs.setdefault('color', 'debug')
|
||
|
|
log(message, *args, **kwargs)
|
||
|
|
|
||
|
|
def killpg(pid):
|
||
|
|
if not hasattr(os, 'killpg'):
|
||
|
|
return
|
||
|
|
try:
|
||
|
|
return os.killpg(pid, 9)
|
||
|
|
except OSError as ex:
|
||
|
|
if ex.errno != 3:
|
||
|
|
log('killpg(%r, 9) failed: %s: %s', pid, type(ex).__name__, ex)
|
||
|
|
except Exception as ex:
|
||
|
|
log('killpg(%r, 9) failed: %s: %s', pid, type(ex).__name__, ex)
|
||
|
|
|
||
|
|
|
||
|
|
def kill_processtree(pid):
|
||
|
|
ignore_msg = 'ERROR: The process "%s" not found.' % pid
|
||
|
|
err = Popen('taskkill /F /PID %s /T' % pid, stderr=subprocess.PIPE).communicate()[1]
|
||
|
|
if err and err.strip() not in [ignore_msg, '']:
|
||
|
|
log('%r', err)
|
||
|
|
|
||
|
|
|
||
|
|
def _kill(popen):
|
||
|
|
if hasattr(popen, 'kill'):
|
||
|
|
try:
|
||
|
|
popen.kill()
|
||
|
|
except OSError as ex:
|
||
|
|
if ex.errno == 3: # No such process
|
||
|
|
return
|
||
|
|
if ex.errno == 13: # Permission denied (translated from windows error 5: "Access is denied")
|
||
|
|
return
|
||
|
|
raise
|
||
|
|
else:
|
||
|
|
try:
|
||
|
|
os.kill(popen.pid, 9)
|
||
|
|
except EnvironmentError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def kill(popen):
|
||
|
|
if popen.timer is not None:
|
||
|
|
popen.timer.cancel()
|
||
|
|
popen.timer = None
|
||
|
|
if popen.poll() is not None:
|
||
|
|
return
|
||
|
|
popen.was_killed = True
|
||
|
|
try:
|
||
|
|
if getattr(popen, 'setpgrp_enabled', None):
|
||
|
|
killpg(popen.pid)
|
||
|
|
elif sys.platform.startswith('win'):
|
||
|
|
kill_processtree(popen.pid)
|
||
|
|
except Exception:
|
||
|
|
traceback.print_exc()
|
||
|
|
try:
|
||
|
|
_kill(popen)
|
||
|
|
except Exception:
|
||
|
|
traceback.print_exc()
|
||
|
|
try:
|
||
|
|
popen.wait()
|
||
|
|
except Exception:
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
# A set of environment keys we ignore for printing purposes
|
||
|
|
IGNORED_GEVENT_ENV_KEYS = {
|
||
|
|
'GEVENTTEST_QUIET',
|
||
|
|
'GEVENT_DEBUG',
|
||
|
|
'GEVENTSETUP_EV_VERIFY',
|
||
|
|
'GEVENTSETUP_EMBED',
|
||
|
|
}
|
||
|
|
|
||
|
|
# A set of (name, value) pairs we ignore for printing purposes.
|
||
|
|
# These should match the defaults.
|
||
|
|
IGNORED_GEVENT_ENV_ITEMS = {
|
||
|
|
('GEVENT_RESOLVER', 'thread'),
|
||
|
|
('GEVENT_RESOLVER_NAMESERVERS', '8.8.8.8'),
|
||
|
|
('GEVENTTEST_USE_RESOURCES', 'all'),
|
||
|
|
}
|
||
|
|
|
||
|
|
def getname(command, env=None, setenv=None):
|
||
|
|
result = []
|
||
|
|
|
||
|
|
env = (env or os.environ).copy()
|
||
|
|
env.update(setenv or {})
|
||
|
|
|
||
|
|
for key, value in sorted(env.items()):
|
||
|
|
if not key.startswith('GEVENT'):
|
||
|
|
continue
|
||
|
|
if key in IGNORED_GEVENT_ENV_KEYS:
|
||
|
|
continue
|
||
|
|
if (key, value) in IGNORED_GEVENT_ENV_ITEMS:
|
||
|
|
continue
|
||
|
|
result.append('%s=%s' % (key, value))
|
||
|
|
|
||
|
|
if isinstance(command, six.string_types):
|
||
|
|
result.append(command)
|
||
|
|
else:
|
||
|
|
result.extend(command)
|
||
|
|
|
||
|
|
return ' '.join(result)
|
||
|
|
|
||
|
|
|
||
|
|
def start(command, quiet=False, **kwargs):
|
||
|
|
timeout = kwargs.pop('timeout', None)
|
||
|
|
preexec_fn = None
|
||
|
|
if not os.environ.get('DO_NOT_SETPGRP'):
|
||
|
|
preexec_fn = getattr(os, 'setpgrp', None)
|
||
|
|
env = kwargs.pop('env', None)
|
||
|
|
setenv = kwargs.pop('setenv', None) or {}
|
||
|
|
name = getname(command, env=env, setenv=setenv)
|
||
|
|
if preexec_fn is not None:
|
||
|
|
setenv['DO_NOT_SETPGRP'] = '1'
|
||
|
|
if setenv:
|
||
|
|
env = env.copy() if env else os.environ.copy()
|
||
|
|
env.update(setenv)
|
||
|
|
|
||
|
|
if not quiet:
|
||
|
|
log('+ %s', name)
|
||
|
|
popen = Popen(command, preexec_fn=preexec_fn, env=env, **kwargs)
|
||
|
|
popen.name = name
|
||
|
|
popen.setpgrp_enabled = preexec_fn is not None
|
||
|
|
popen.was_killed = False
|
||
|
|
if timeout is not None:
|
||
|
|
t = get_original('threading', 'Timer')(timeout, kill, args=(popen, ))
|
||
|
|
popen.timer = t
|
||
|
|
t.daemon = True
|
||
|
|
t.start()
|
||
|
|
popen.timer = t
|
||
|
|
return popen
|
||
|
|
|
||
|
|
|
||
|
|
class RunResult(object):
|
||
|
|
"""
|
||
|
|
The results of running an external command.
|
||
|
|
|
||
|
|
If the command was successful, this has a boolean
|
||
|
|
value of True; otherwise, a boolean value of false.
|
||
|
|
|
||
|
|
The integer value of this object is the command's exit code.
|
||
|
|
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self,
|
||
|
|
command,
|
||
|
|
run_kwargs,
|
||
|
|
code,
|
||
|
|
output=None, # type: str
|
||
|
|
error=None, # type: str
|
||
|
|
name=None,
|
||
|
|
run_count=0, skipped_count=0,
|
||
|
|
run_duration=0, # type: float
|
||
|
|
):
|
||
|
|
self.command = command
|
||
|
|
self.run_kwargs = run_kwargs
|
||
|
|
self.code = code
|
||
|
|
self.output = output
|
||
|
|
self.error = error
|
||
|
|
self.name = name
|
||
|
|
self.run_count = run_count
|
||
|
|
self.skipped_count = skipped_count
|
||
|
|
self.run_duration = run_duration
|
||
|
|
|
||
|
|
@property
|
||
|
|
def output_lines(self):
|
||
|
|
return self.output.splitlines()
|
||
|
|
|
||
|
|
def __bool__(self):
|
||
|
|
return not bool(self.code)
|
||
|
|
|
||
|
|
__nonzero__ = __bool__
|
||
|
|
|
||
|
|
def __int__(self):
|
||
|
|
return self.code
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
return (
|
||
|
|
"RunResult of: %r\n"
|
||
|
|
"Code: %s\n"
|
||
|
|
"kwargs: %r\n"
|
||
|
|
"Output:\n"
|
||
|
|
"----\n"
|
||
|
|
"%s"
|
||
|
|
"----\n"
|
||
|
|
"Error:\n"
|
||
|
|
"----\n"
|
||
|
|
"%s"
|
||
|
|
"----\n"
|
||
|
|
) % (
|
||
|
|
self.command,
|
||
|
|
self.code,
|
||
|
|
self.run_kwargs,
|
||
|
|
self.output,
|
||
|
|
self.error
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _should_show_warning_output(out):
|
||
|
|
if 'Warning' in out:
|
||
|
|
# Strip out some patterns we specifically do not
|
||
|
|
# care about.
|
||
|
|
# from test.support for monkey-patched tests
|
||
|
|
out = out.replace('Warning -- reap_children', 'NADA')
|
||
|
|
out = out.replace("Warning -- threading_cleanup", 'NADA')
|
||
|
|
|
||
|
|
# The below *could* be done with sophisticated enough warning
|
||
|
|
# filters passed to the children
|
||
|
|
|
||
|
|
# collections.abc is the new home; setuptools uses the old one,
|
||
|
|
# as does dnspython
|
||
|
|
out = out.replace("DeprecationWarning: Using or importing the ABCs", 'NADA')
|
||
|
|
# libuv poor timer resolution
|
||
|
|
out = out.replace('UserWarning: libuv only supports', 'NADA')
|
||
|
|
# Packages on Python 2
|
||
|
|
out = out.replace('ImportWarning: Not importing directory', 'NADA')
|
||
|
|
# Testing that U mode does the same thing
|
||
|
|
out = out.replace("DeprecationWarning: 'U' mode is deprecated", 'NADA')
|
||
|
|
out = out.replace("DeprecationWarning: dns.hash module", 'NADA')
|
||
|
|
return 'Warning' in out
|
||
|
|
|
||
|
|
output_lock = threading.Lock()
|
||
|
|
|
||
|
|
def _find_test_status(took, out):
|
||
|
|
status = '[took %.1fs%s]'
|
||
|
|
skipped = ''
|
||
|
|
run_count = 0
|
||
|
|
skipped_count = 0
|
||
|
|
if out:
|
||
|
|
m = re.search(r"Ran (\d+) tests in", out)
|
||
|
|
if m:
|
||
|
|
result = out[m.start():m.end()]
|
||
|
|
status = status.replace('took', result)
|
||
|
|
run_count = int(out[m.start(1):m.end(1)])
|
||
|
|
|
||
|
|
m = re.search(r' \(skipped=(\d+)\)$', out)
|
||
|
|
if m:
|
||
|
|
skipped = _colorize('skipped', out[m.start():m.end()])
|
||
|
|
skipped_count = int(out[m.start(1):m.end(1)])
|
||
|
|
status = status % (took, skipped) # pylint:disable=consider-using-augmented-assign
|
||
|
|
if took > 10:
|
||
|
|
status = _colorize('slow-test', status)
|
||
|
|
return status, run_count, skipped_count
|
||
|
|
|
||
|
|
|
||
|
|
def run(command, **kwargs): # pylint:disable=too-many-locals
|
||
|
|
"""
|
||
|
|
Execute *command*, returning a `RunResult`.
|
||
|
|
|
||
|
|
This blocks until *command* finishes or until it times out.
|
||
|
|
"""
|
||
|
|
buffer_output = kwargs.pop('buffer_output', BUFFER_OUTPUT)
|
||
|
|
quiet = kwargs.pop('quiet', QUIET)
|
||
|
|
verbose = not quiet
|
||
|
|
nested = kwargs.pop('nested', False)
|
||
|
|
if buffer_output:
|
||
|
|
assert 'stdout' not in kwargs and 'stderr' not in kwargs, kwargs
|
||
|
|
kwargs['stderr'] = subprocess.STDOUT
|
||
|
|
kwargs['stdout'] = subprocess.PIPE
|
||
|
|
popen = start(command, quiet=quiet, **kwargs)
|
||
|
|
name = popen.name
|
||
|
|
|
||
|
|
try:
|
||
|
|
time_start = perf_counter()
|
||
|
|
out, err = popen.communicate()
|
||
|
|
duration = perf_counter() - time_start
|
||
|
|
if popen.was_killed or popen.poll() is None:
|
||
|
|
result = 'TIMEOUT'
|
||
|
|
else:
|
||
|
|
result = popen.poll()
|
||
|
|
finally:
|
||
|
|
kill(popen)
|
||
|
|
assert popen.timer is None
|
||
|
|
|
||
|
|
|
||
|
|
failed = bool(result)
|
||
|
|
if out:
|
||
|
|
out = out.strip()
|
||
|
|
out = out if isinstance(out, str) else out.decode('utf-8', 'ignore')
|
||
|
|
if out and (failed or verbose or _should_show_warning_output(out)):
|
||
|
|
out = ' ' + out.replace('\n', '\n ')
|
||
|
|
out = out.rstrip()
|
||
|
|
out += '\n'
|
||
|
|
log('| %s\n%s', name, out)
|
||
|
|
status, run_count, skipped_count = _find_test_status(duration, out)
|
||
|
|
if result:
|
||
|
|
log('! %s [code %s] %s', name, result, status, color='error')
|
||
|
|
elif not nested:
|
||
|
|
log('- %s %s', name, status)
|
||
|
|
return RunResult(
|
||
|
|
command, kwargs, result,
|
||
|
|
output=out, error=err,
|
||
|
|
name=name,
|
||
|
|
run_count=run_count,
|
||
|
|
skipped_count=skipped_count,
|
||
|
|
run_duration=duration,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class NoSetupPyFound(Exception):
|
||
|
|
"Raised by find_setup_py_above"
|
||
|
|
|
||
|
|
def find_setup_py_above(a_file):
|
||
|
|
"Return the directory containing setup.py somewhere above *a_file*"
|
||
|
|
root = os.path.dirname(os.path.abspath(a_file))
|
||
|
|
while not os.path.exists(os.path.join(root, 'setup.py')):
|
||
|
|
prev, root = root, os.path.dirname(root)
|
||
|
|
if root == prev:
|
||
|
|
# Let's avoid infinite loops at root
|
||
|
|
raise NoSetupPyFound('could not find my setup.py above %r' % (a_file,))
|
||
|
|
return root
|
||
|
|
|
||
|
|
def search_for_setup_py(a_file=None, a_module_name=None, a_class=None, climb_cwd=True):
|
||
|
|
if a_file is not None:
|
||
|
|
try:
|
||
|
|
return find_setup_py_above(a_file)
|
||
|
|
except NoSetupPyFound:
|
||
|
|
pass
|
||
|
|
|
||
|
|
if a_class is not None:
|
||
|
|
try:
|
||
|
|
return find_setup_py_above(sys.modules[a_class.__module__].__file__)
|
||
|
|
except NoSetupPyFound:
|
||
|
|
pass
|
||
|
|
|
||
|
|
if a_module_name is not None:
|
||
|
|
try:
|
||
|
|
return find_setup_py_above(sys.modules[a_module_name].__file__)
|
||
|
|
except NoSetupPyFound:
|
||
|
|
pass
|
||
|
|
|
||
|
|
if climb_cwd:
|
||
|
|
return find_setup_py_above("./dne")
|
||
|
|
|
||
|
|
raise NoSetupPyFound("After checking %r" % (locals(),))
|
||
|
|
|
||
|
|
def _version_dir_components():
|
||
|
|
directory = '%s.%s' % sys.version_info[:2]
|
||
|
|
full_directory = '%s.%s.%s' % sys.version_info[:3]
|
||
|
|
if hasattr(sys, 'pypy_version_info'):
|
||
|
|
directory += 'pypy'
|
||
|
|
full_directory += 'pypy'
|
||
|
|
|
||
|
|
return directory, full_directory
|
||
|
|
|
||
|
|
def find_stdlib_tests():
|
||
|
|
"""
|
||
|
|
Return a sequence of directories that could contain
|
||
|
|
stdlib tests for the running version of Python.
|
||
|
|
|
||
|
|
The most specific tests are at the end of the sequence.
|
||
|
|
|
||
|
|
No checks are performed on existence of the directories.
|
||
|
|
"""
|
||
|
|
setup_py = search_for_setup_py(a_file=__file__)
|
||
|
|
greentest = os.path.join(setup_py, 'src', 'greentest')
|
||
|
|
|
||
|
|
|
||
|
|
directory, full_directory = _version_dir_components()
|
||
|
|
|
||
|
|
directory = '%s.%s' % sys.version_info[:2]
|
||
|
|
full_directory = '%s.%s.%s' % sys.version_info[:3]
|
||
|
|
if hasattr(sys, 'pypy_version_info'):
|
||
|
|
directory += 'pypy'
|
||
|
|
full_directory += 'pypy'
|
||
|
|
|
||
|
|
directory = os.path.join(greentest, directory)
|
||
|
|
full_directory = os.path.join(greentest, full_directory)
|
||
|
|
|
||
|
|
return directory, full_directory
|
||
|
|
|
||
|
|
def absolute_pythonpath():
|
||
|
|
"""
|
||
|
|
Return the PYTHONPATH environment variable (if set) with each
|
||
|
|
entry being an absolute path. If not set, returns None.
|
||
|
|
"""
|
||
|
|
if 'PYTHONPATH' not in os.environ:
|
||
|
|
return None
|
||
|
|
|
||
|
|
path = os.environ['PYTHONPATH']
|
||
|
|
path = [os.path.abspath(p) for p in path.split(os.path.pathsep)]
|
||
|
|
return os.path.pathsep.join(path)
|
||
|
|
|
||
|
|
class ExampleMixin(object):
|
||
|
|
"""
|
||
|
|
Something that uses the ``examples/`` directory
|
||
|
|
from the root of the gevent distribution.
|
||
|
|
|
||
|
|
The `cwd` property is set to the root of the gevent distribution.
|
||
|
|
"""
|
||
|
|
#: Arguments to pass to the example file.
|
||
|
|
example_args = []
|
||
|
|
before_delay = 3
|
||
|
|
after_delay = 0.5
|
||
|
|
#: Path of the example Python file, relative to `cwd`
|
||
|
|
example = None # subclasses define this to be the path to the server.py
|
||
|
|
#: Keyword arguments to pass to the start or run method.
|
||
|
|
start_kwargs = None
|
||
|
|
|
||
|
|
def find_setup_py(self):
|
||
|
|
"Return the directory containing setup.py"
|
||
|
|
return search_for_setup_py(
|
||
|
|
a_file=__file__,
|
||
|
|
a_class=type(self)
|
||
|
|
)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def cwd(self):
|
||
|
|
try:
|
||
|
|
root = self.find_setup_py()
|
||
|
|
except NoSetupPyFound as e:
|
||
|
|
raise unittest.SkipTest("Unable to locate file/dir to run: %s" % (e,))
|
||
|
|
return os.path.join(root, 'examples')
|
||
|
|
|
||
|
|
@property
|
||
|
|
def setenv(self):
|
||
|
|
"""
|
||
|
|
Returns a dictionary of environment variables to set for the
|
||
|
|
child in addition to (or replacing) the ones already in the
|
||
|
|
environment.
|
||
|
|
|
||
|
|
Since the child is run in `cwd`, relative paths in ``PYTHONPATH``
|
||
|
|
need to be converted to absolute paths.
|
||
|
|
"""
|
||
|
|
abs_pythonpath = absolute_pythonpath()
|
||
|
|
return {'PYTHONPATH': abs_pythonpath} if abs_pythonpath else None
|
||
|
|
|
||
|
|
def _start(self, meth):
|
||
|
|
if getattr(self, 'args', None):
|
||
|
|
raise AssertionError("Invalid test", self, self.args)
|
||
|
|
if getattr(self, 'server', None):
|
||
|
|
raise AssertionError("Invalid test", self, self.server)
|
||
|
|
|
||
|
|
try:
|
||
|
|
# These could be or are properties that can raise
|
||
|
|
server = self.example
|
||
|
|
server_dir = self.cwd
|
||
|
|
except NoSetupPyFound as e:
|
||
|
|
raise unittest.SkipTest("Unable to locate file/dir to run: %s" % (e,))
|
||
|
|
|
||
|
|
kwargs = self.start_kwargs or {}
|
||
|
|
setenv = self.setenv
|
||
|
|
if setenv:
|
||
|
|
if 'setenv' in kwargs:
|
||
|
|
kwargs['setenv'].update(setenv)
|
||
|
|
else:
|
||
|
|
kwargs['setenv'] = setenv
|
||
|
|
return meth(
|
||
|
|
[sys.executable, '-W', 'ignore', '-u', server] + self.example_args,
|
||
|
|
cwd=server_dir,
|
||
|
|
**kwargs
|
||
|
|
)
|
||
|
|
|
||
|
|
def start_example(self):
|
||
|
|
return self._start(meth=start)
|
||
|
|
|
||
|
|
def run_example(self):# run() is a unittest method.
|
||
|
|
return self._start(meth=run)
|
||
|
|
|
||
|
|
|
||
|
|
class TestServer(ExampleMixin,
|
||
|
|
unittest.TestCase):
|
||
|
|
popen = None
|
||
|
|
|
||
|
|
def running_server(self):
|
||
|
|
from contextlib import contextmanager
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def running_server():
|
||
|
|
with self.start_example() as popen:
|
||
|
|
self.popen = popen
|
||
|
|
self.before()
|
||
|
|
yield
|
||
|
|
self.after()
|
||
|
|
return running_server()
|
||
|
|
|
||
|
|
def test(self):
|
||
|
|
with self.running_server():
|
||
|
|
self._run_all_tests()
|
||
|
|
|
||
|
|
def before(self):
|
||
|
|
if self.before_delay is not None:
|
||
|
|
sleep(self.before_delay)
|
||
|
|
self.assertIsNone(self.popen.poll(),
|
||
|
|
'%s died with code %s' % (
|
||
|
|
self.example, self.popen.poll(),
|
||
|
|
))
|
||
|
|
|
||
|
|
def after(self):
|
||
|
|
if self.after_delay is not None:
|
||
|
|
sleep(self.after_delay)
|
||
|
|
self.assertIsNone(self.popen.poll(),
|
||
|
|
'%s died with code %s' % (
|
||
|
|
self.example, self.popen.poll(),
|
||
|
|
))
|
||
|
|
|
||
|
|
def _run_all_tests(self):
|
||
|
|
ran = False
|
||
|
|
for method in sorted(dir(self)):
|
||
|
|
if method.startswith('_test'):
|
||
|
|
function = getattr(self, method)
|
||
|
|
if callable(function):
|
||
|
|
function()
|
||
|
|
ran = True
|
||
|
|
assert ran
|
||
|
|
|
||
|
|
|
||
|
|
class alarm(threading.Thread):
|
||
|
|
# can't use signal.alarm because of Windows
|
||
|
|
|
||
|
|
def __init__(self, timeout):
|
||
|
|
threading.Thread.__init__(self)
|
||
|
|
self.daemon = True
|
||
|
|
self.timeout = timeout
|
||
|
|
self.start()
|
||
|
|
|
||
|
|
def run(self):
|
||
|
|
sleep(self.timeout)
|
||
|
|
sys.stderr.write('Timeout.\n')
|
||
|
|
os._exit(5)
|