- refine lsof checking

- make runpytest() create an inline testing process instead of
  a subprocess one

--HG--
branch : testrefactor
This commit is contained in:
holger krekel 2015-04-28 11:54:46 +02:00
parent d3e363b97a
commit a8afba054a
18 changed files with 137 additions and 116 deletions

View File

@ -38,8 +38,11 @@ def main(args=None, plugins=None):
tw.line("ERROR: could not load %s\n" % (e.path), red=True)
return 4
else:
config.pluginmanager.check_pending()
return config.hook.pytest_cmdline_main(config=config)
try:
config.pluginmanager.check_pending()
return config.hook.pytest_cmdline_main(config=config)
finally:
config._ensure_unconfigure()
except UsageError as e:
for msg in e.args:
sys.stderr.write("ERROR: %s\n" %(msg,))
@ -85,12 +88,18 @@ def _prepareconfig(args=None, plugins=None):
if not isinstance(args, str):
raise ValueError("not a string or argument list: %r" % (args,))
args = shlex.split(args)
pluginmanager = get_config().pluginmanager
if plugins:
for plugin in plugins:
pluginmanager.register(plugin)
return pluginmanager.hook.pytest_cmdline_parse(
pluginmanager=pluginmanager, args=args)
config = get_config()
pluginmanager = config.pluginmanager
try:
if plugins:
for plugin in plugins:
pluginmanager.register(plugin)
return pluginmanager.hook.pytest_cmdline_parse(
pluginmanager=pluginmanager, args=args)
except BaseException:
config._ensure_unconfigure()
raise
def exclude_pytest_names(name):
return not name.startswith(name) or name == "pytest_plugins" or \
@ -263,7 +272,10 @@ class PytestPluginManager(PluginManager):
def consider_pluginarg(self, arg):
if arg.startswith("no:"):
self.set_blocked(arg[3:])
name = arg[3:]
self.set_blocked(name)
if not name.startswith("pytest_"):
self.set_blocked("pytest_" + name)
else:
self.import_plugin(arg)

View File

@ -1,5 +1,6 @@
""" (disabled by default) support for testing pytest and pytest plugins. """
import sys
import traceback
import os
import codecs
import re
@ -287,7 +288,8 @@ class TmpTestdir:
break
self.tmpdir = tmpdir
self.plugins = []
self._savesyspath = list(sys.path)
self._savesyspath = (list(sys.path), list(sys.meta_path))
self._savemodulekeys = set(sys.modules)
self.chdir() # always chdir
self.request.addfinalizer(self.finalize)
@ -303,23 +305,23 @@ class TmpTestdir:
has finished.
"""
sys.path[:] = self._savesyspath
sys.path[:], sys.meta_path[:] = self._savesyspath
if hasattr(self, '_olddir'):
self._olddir.chdir()
self.delete_loaded_modules()
def delete_loaded_modules(self):
"""Delete modules that have been loaded from tmpdir.
"""Delete modules that have been loaded during a test.
This allows the interpreter to catch module changes in case
the module is re-imported.
"""
for name, mod in list(sys.modules.items()):
if mod:
fn = getattr(mod, '__file__', None)
if fn and fn.startswith(str(self.tmpdir)):
del sys.modules[name]
for name in set(sys.modules).difference(self._savemodulekeys):
# it seems zope.interfaces is keeping some state
# (used by twisted related tests)
if name != "zope.interface":
del sys.modules[name]
def make_hook_recorder(self, pluginmanager):
"""Create a new :py:class:`HookRecorder` for a PluginManager."""
@ -584,16 +586,27 @@ class TmpTestdir:
reprec.ret = ret
return reprec
def inline_runpytest(self, *args):
def inline_runpytest(self, *args, **kwargs):
""" Return result of running pytest in-process, providing a similar
interface to what self.runpytest() provides. """
if kwargs.get("syspathinsert"):
self.syspathinsert()
now = time.time()
capture = py.io.StdCaptureFD()
capture = py.io.StdCapture()
try:
reprec = self.inline_run(*args)
try:
reprec = self.inline_run(*args)
except SystemExit as e:
class reprec:
ret = e.args[0]
except Exception:
traceback.print_exc()
class reprec:
ret = 3
finally:
out, err = capture.reset()
assert out or err
sys.stdout.write(out)
sys.stderr.write(err)
res = RunResult(reprec.ret,
out.split("\n"), err.split("\n"),
@ -601,6 +614,9 @@ class TmpTestdir:
res.reprec = reprec
return res
def runpytest(self, *args, **kwargs):
return self.inline_runpytest(*args, **kwargs)
def parseconfig(self, *args):
"""Return a new py.test Config instance from given commandline args.
@ -822,7 +838,7 @@ class TmpTestdir:
command = self._getsysprepend() + command
return self.run(sys.executable, "-c", command)
def runpytest(self, *args):
def runpytest_subprocess(self, *args):
"""Run py.test as a subprocess with given arguments.
Any plugins added to the :py:attr:`plugins` list will added

View File

@ -7,7 +7,7 @@ def test_failure_demo_fails_properly(testdir):
target = testdir.tmpdir.join(failure_demo.basename)
failure_demo.copy(target)
failure_demo.copy(testdir.tmpdir.join(failure_demo.basename))
result = testdir.runpytest(target)
result = testdir.runpytest(target, syspathinsert=True)
result.stdout.fnmatch_lines([
"*42 failed*"
])

View File

@ -82,7 +82,7 @@ class TestGeneralUsage:
def test_option(pytestconfig):
assert pytestconfig.option.xyz == "123"
""")
result = testdir.runpytest("-p", "pytest_xyz", "--xyz=123")
result = testdir.runpytest("-p", "pytest_xyz", "--xyz=123", syspathinsert=True)
assert result.ret == 0
result.stdout.fnmatch_lines([
'*1 passed*',
@ -353,7 +353,8 @@ class TestGeneralUsage:
*unrecognized*
""")
def test_getsourcelines_error_issue553(self, testdir):
def test_getsourcelines_error_issue553(self, testdir, monkeypatch):
monkeypatch.setattr("inspect.getsourcelines", None)
p = testdir.makepyfile("""
def raise_error(obj):
raise IOError('source code not available')

View File

@ -1,5 +1,6 @@
import pytest
import sys
import gc
pytest_plugins = "pytester",
@ -17,8 +18,8 @@ class LsofFdLeakChecker(object):
def _parse_lsof_output(self, out):
def isopen(line):
return line.startswith('f') and (
"deleted" not in line and 'mem' not in line and "txt" not in line and 'cwd' not in line)
return line.startswith('f') and ("deleted" not in line and
'mem' not in line and "txt" not in line and 'cwd' not in line)
open_files = []
@ -32,46 +33,49 @@ class LsofFdLeakChecker(object):
return open_files
def matching_platform(self):
try:
py.process.cmdexec("lsof -v")
except py.process.cmdexec.Error:
return False
else:
return True
@pytest.hookimpl_opts(hookwrapper=True, tryfirst=True)
def pytest_runtest_item(self, item):
lines1 = self.get_open_files()
yield
if hasattr(sys, "pypy_version_info"):
gc.collect()
lines2 = self.get_open_files()
new_fds = set([t[0] for t in lines2]) - set([t[0] for t in lines1])
leaked_files = [t for t in lines2 if t[0] in new_fds]
if leaked_files:
error = []
error.append("***** %s FD leakage detected" % len(leaked_files))
error.extend([str(f) for f in leaked_files])
error.append("*** Before:")
error.extend([str(f) for f in lines1])
error.append("*** After:")
error.extend([str(f) for f in lines2])
error.append(error[0])
error.append("*** function %s:%s: %s " % item.location)
pytest.fail("\n".join(error), pytrace=False)
def pytest_addoption(parser):
parser.addoption('--lsof',
action="store_true", dest="lsof", default=False,
help=("run FD checks if lsof is available"))
def pytest_runtest_setup(item):
config = item.config
config._basedir = py.path.local()
def pytest_configure(config):
if config.getvalue("lsof"):
try:
config._fd_leak_checker = LsofFdLeakChecker()
config._openfiles = config._fd_leak_checker.get_open_files()
except py.process.cmdexec.Error:
pass
checker = LsofFdLeakChecker()
if checker.matching_platform():
config.pluginmanager.register(checker)
#def pytest_report_header():
# return "pid: %s" % os.getpid()
def check_open_files(config):
lines2 = config._fd_leak_checker.get_open_files()
new_fds = set([t[0] for t in lines2]) - set([t[0] for t in config._openfiles])
open_files = [t for t in lines2 if t[0] in new_fds]
if open_files:
error = []
error.append("***** %s FD leakage detected" % len(open_files))
error.extend([str(f) for f in open_files])
error.append("*** Before:")
error.extend([str(f) for f in config._openfiles])
error.append("*** After:")
error.extend([str(f) for f in lines2])
error.append(error[0])
raise AssertionError("\n".join(error))
@pytest.hookimpl_opts(hookwrapper=True, trylast=True)
def pytest_runtest_teardown(item):
yield
item.config._basedir.chdir()
if hasattr(item.config, '_openfiles'):
check_open_files(item.config)
# XXX copied from execnet's conftest.py - needs to be merged
winpymap = {

View File

@ -626,10 +626,8 @@ def test_setup_only_available_in_subdir(testdir):
"""))
sub1.join("test_in_sub1.py").write("def test_1(): pass")
sub2.join("test_in_sub2.py").write("def test_2(): pass")
result = testdir.inline_runpytest("-v", "-s")
result.stdout.fnmatch_lines([
"*2 passed*"
])
result = testdir.runpytest("-v", "-s")
result.assert_outcomes(passed=2)
def test_modulecol_roundtrip(testdir):
modcol = testdir.getmodulecol("pass", withinit=True)

View File

@ -99,10 +99,8 @@ class TestFillFixtures:
sub1.join("test_in_sub1.py").write("def test_1(arg1): pass")
sub2.join("test_in_sub2.py").write("def test_2(arg2): pass")
result = testdir.inline_runpytest("-v")
result.stdout.fnmatch_lines([
"*2 passed*"
])
result = testdir.runpytest("-v")
result.assert_outcomes(passed=2)
def test_extend_fixture_module_class(self, testdir):
testfile = testdir.makepyfile("""

View File

@ -663,7 +663,7 @@ class TestMetafuncFunctional:
"""))
sub1.join("test_in_sub1.py").write("def test_1(): pass")
sub2.join("test_in_sub2.py").write("def test_2(): pass")
result = testdir.inline_runpytest("-v", "-s", sub1, sub2, sub1)
result = testdir.runpytest("-v", "-s", sub1, sub2, sub1)
result.assert_outcomes(passed=3)
def test_generate_same_function_names_issue403(self, testdir):

View File

@ -451,7 +451,7 @@ def test_assertion_options(testdir):
x = 3
assert x == 4
""")
result = testdir.inline_runpytest()
result = testdir.runpytest()
assert "3 == 4" in result.stdout.str()
off_options = (("--no-assert",),
("--nomagic",),
@ -461,7 +461,7 @@ def test_assertion_options(testdir):
("--assert=plain", "--nomagic"),
("--assert=plain", "--no-assert", "--nomagic"))
for opt in off_options:
result = testdir.runpytest(*opt)
result = testdir.runpytest_subprocess(*opt)
assert "3 == 4" not in result.stdout.str()
def test_old_assert_mode(testdir):
@ -469,7 +469,7 @@ def test_old_assert_mode(testdir):
def test_in_old_mode():
assert "@py_builtins" not in globals()
""")
result = testdir.runpytest("--assert=reinterp")
result = testdir.runpytest_subprocess("--assert=reinterp")
assert result.ret == 0
def test_triple_quoted_string_issue113(testdir):

View File

@ -453,7 +453,7 @@ def test_rewritten():
assert not os.path.exists(__cached__)
assert not os.path.exists(os.path.dirname(__cached__))""")
monkeypatch.setenv("PYTHONDONTWRITEBYTECODE", "1")
assert testdir.runpytest().ret == 0
assert testdir.runpytest_subprocess().ret == 0
@pytest.mark.skipif('"__pypy__" in sys.modules')
def test_pyc_vs_pyo(self, testdir, monkeypatch):
@ -615,10 +615,8 @@ class TestAssertionRewriteHookDetails(object):
testdir.makepyfile(**contents)
testdir.maketxtfile(**{'testpkg/resource': "Load me please."})
result = testdir.runpytest()
result.stdout.fnmatch_lines([
'* 1 passed*',
])
result = testdir.runpytest_subprocess()
result.assert_outcomes(passed=1)
def test_read_pyc(self, tmpdir):
"""

View File

@ -121,7 +121,7 @@ def test_capturing_unicode(testdir, method):
print (sys.stdout)
print (%s)
""" % obj)
result = testdir.runpytest("--capture=%s" % method)
result = testdir.runpytest_subprocess("--capture=%s" % method)
result.stdout.fnmatch_lines([
"*1 passed*"
])
@ -133,7 +133,7 @@ def test_capturing_bytes_in_utf8_encoding(testdir, method):
def test_unicode():
print ('b\\u00f6y')
""")
result = testdir.runpytest("--capture=%s" % method)
result = testdir.runpytest_subprocess("--capture=%s" % method)
result.stdout.fnmatch_lines([
"*1 passed*"
])
@ -144,7 +144,7 @@ def test_collect_capturing(testdir):
print ("collect %s failure" % 13)
import xyz42123
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"*Captured stdout*",
"*collect 13 failure*",
@ -165,7 +165,7 @@ class TestPerTestCapturing:
print ("in func2")
assert 0
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"setup module*",
"setup test_func1*",
@ -188,7 +188,7 @@ class TestPerTestCapturing:
def teardown_function(func):
print ("in teardown")
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"*test_func():*",
"*Captured stdout during setup*",
@ -206,7 +206,7 @@ class TestPerTestCapturing:
print ("in func2")
assert 0
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
s = result.stdout.str()
assert "in func1" not in s
assert "in func2" in s
@ -222,7 +222,7 @@ class TestPerTestCapturing:
print ("in func1")
pass
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
'*teardown_function*',
'*Captured stdout*',
@ -240,7 +240,7 @@ class TestPerTestCapturing:
def test_func():
pass
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"*def teardown_module(mod):*",
"*Captured stdout*",
@ -259,7 +259,7 @@ class TestPerTestCapturing:
sys.stderr.write(str(2))
raise ValueError
""")
result = testdir.runpytest(p1)
result = testdir.runpytest_subprocess(p1)
result.stdout.fnmatch_lines([
"*test_capturing_outerr.py .F",
"====* FAILURES *====",
@ -282,7 +282,7 @@ class TestLoggingInteraction:
logging.basicConfig(stream=stream)
stream.close() # to free memory/release resources
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stderr.str().find("atexit") == -1
def test_logging_and_immediate_setupteardown(self, testdir):
@ -301,7 +301,7 @@ class TestLoggingInteraction:
""")
for optargs in (('--capture=sys',), ('--capture=fd',)):
print (optargs)
result = testdir.runpytest(p, *optargs)
result = testdir.runpytest_subprocess(p, *optargs)
s = result.stdout.str()
result.stdout.fnmatch_lines([
"*WARN*hello3", # errors show first!
@ -327,7 +327,7 @@ class TestLoggingInteraction:
""")
for optargs in (('--capture=sys',), ('--capture=fd',)):
print (optargs)
result = testdir.runpytest(p, *optargs)
result = testdir.runpytest_subprocess(p, *optargs)
s = result.stdout.str()
result.stdout.fnmatch_lines([
"*WARN*hello3", # errors come first
@ -348,7 +348,7 @@ class TestLoggingInteraction:
logging.warn("hello432")
assert 0
""")
result = testdir.runpytest(
result = testdir.runpytest_subprocess(
p, "--traceconfig",
"-p", "no:capturelog")
assert result.ret != 0
@ -364,7 +364,7 @@ class TestLoggingInteraction:
logging.warn("hello435")
""")
# make sure that logging is still captured in tests
result = testdir.runpytest("-s", "-p", "no:capturelog")
result = testdir.runpytest_subprocess("-s", "-p", "no:capturelog")
assert result.ret == 0
result.stderr.fnmatch_lines([
"WARNING*hello435*",
@ -383,7 +383,7 @@ class TestLoggingInteraction:
logging.warn("hello433")
assert 0
""")
result = testdir.runpytest(p, "-p", "no:capturelog")
result = testdir.runpytest_subprocess(p, "-p", "no:capturelog")
assert result.ret != 0
result.stdout.fnmatch_lines([
"WARNING*hello433*",
@ -410,7 +410,7 @@ class TestCaptureFixture:
def test_two(capfd, capsys):
pass
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"*ERROR*setup*test_one*",
"*capsys*capfd*same*time*",
@ -425,7 +425,7 @@ class TestCaptureFixture:
print ("xxx42xxx")
assert 0
""" % method)
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"xxx42xxx",
])
@ -447,7 +447,7 @@ class TestCaptureFixture:
def test_hello(capsys, missingarg):
pass
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"*test_partial_setup_failure*",
"*1 error*",
@ -461,7 +461,7 @@ class TestCaptureFixture:
os.write(1, str(42).encode('ascii'))
raise KeyboardInterrupt()
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
result.stdout.fnmatch_lines([
"*KeyboardInterrupt*"
])
@ -474,7 +474,7 @@ class TestCaptureFixture:
def test_log(capsys):
logging.error('x')
""")
result = testdir.runpytest(p)
result = testdir.runpytest_subprocess(p)
assert 'closed' not in result.stderr.str()
@ -485,7 +485,7 @@ def test_setup_failure_does_not_kill_capturing(testdir):
raise ValueError(42)
"""))
sub1.join("test_mod.py").write("def test_func1(): pass")
result = testdir.runpytest(testdir.tmpdir, '--traceconfig')
result = testdir.runpytest_subprocess(testdir.tmpdir, '--traceconfig')
result.stdout.fnmatch_lines([
"*ValueError(42)*",
"*1 error*"
@ -500,7 +500,7 @@ def test_fdfuncarg_skips_on_no_osdup(testdir):
def test_hello(capfd):
pass
""")
result = testdir.runpytest("--capture=no")
result = testdir.runpytest_subprocess("--capture=no")
result.stdout.fnmatch_lines([
"*1 skipped*"
])
@ -512,7 +512,7 @@ def test_capture_conftest_runtest_setup(testdir):
print ("hello19")
""")
testdir.makepyfile("def test_func(): pass")
result = testdir.runpytest()
result = testdir.runpytest_subprocess()
assert result.ret == 0
assert 'hello19' not in result.stdout.str()
@ -526,7 +526,7 @@ def test_capture_badoutput_issue412(testdir):
os.write(1, omg)
assert 0
""")
result = testdir.runpytest('--cap=fd')
result = testdir.runpytest_subprocess('--cap=fd')
result.stdout.fnmatch_lines('''
*def test_func*
*assert 0*
@ -541,7 +541,7 @@ def test_capture_early_option_parsing(testdir):
print ("hello19")
""")
testdir.makepyfile("def test_func(): pass")
result = testdir.runpytest("-vs")
result = testdir.runpytest_subprocess("-vs")
assert result.ret == 0
assert 'hello19' in result.stdout.str()
@ -562,7 +562,7 @@ def test_capture_binary_output(testdir):
if __name__ == '__main__':
test_foo()
""")
result = testdir.runpytest('--assert=plain')
result = testdir.runpytest_subprocess('--assert=plain')
result.stdout.fnmatch_lines([
'*2 passed*',
])
@ -885,7 +885,7 @@ class TestStdCaptureFD(TestStdCapture):
os.write(1, "hello\\n".encode("ascii"))
assert 0
""")
result = testdir.runpytest()
result = testdir.runpytest_subprocess()
result.stdout.fnmatch_lines("""
*test_x*
*assert 0*
@ -936,7 +936,7 @@ class TestStdCaptureFDinvalidFD:
cap = StdCaptureFD(out=False, err=False, in_=True)
cap.stop_capturing()
""")
result = testdir.runpytest("--capture=fd")
result = testdir.runpytest_subprocess("--capture=fd")
assert result.ret == 0
assert result.parseoutcomes()['passed'] == 3
@ -971,7 +971,7 @@ def test_close_and_capture_again(testdir):
os.write(1, b"hello\\n")
assert 0
""")
result = testdir.runpytest()
result = testdir.runpytest_subprocess()
result.stdout.fnmatch_lines("""
*test_capture_again*
*assert 0*

View File

@ -296,7 +296,6 @@ class TestSession:
subdir.ensure("__init__.py")
target = subdir.join(p.basename)
p.move(target)
testdir.chdir()
subdir.chdir()
config = testdir.parseconfig(p.basename)
rcol = Session(config=config)
@ -470,7 +469,6 @@ class Test_getinitialnodes:
assert col.config is config
def test_pkgfile(self, testdir):
testdir.chdir()
tmpdir = testdir.tmpdir
subdir = tmpdir.join("subdir")
x = subdir.ensure("x.py")

View File

@ -961,7 +961,7 @@ class TestPytestPluginManager:
""")
p.copy(p.dirpath("skipping2.py"))
monkeypatch.setenv("PYTEST_PLUGINS", "skipping2")
result = testdir.runpytest("-rw", "-p", "skipping1", "--traceconfig")
result = testdir.runpytest("-rw", "-p", "skipping1", syspathinsert=True)
assert result.ret == 0
result.stdout.fnmatch_lines([
"WI1*skipped plugin*skipping1*hello*",
@ -990,7 +990,7 @@ class TestPytestPluginManager:
assert plugin is not None
""")
monkeypatch.setenv('PYTEST_PLUGINS', 'pytest_x500', prepend=",")
result = testdir.runpytest(p)
result = testdir.runpytest(p, syspathinsert=True)
assert result.ret == 0
result.stdout.fnmatch_lines(["*1 passed*"])

View File

@ -75,8 +75,6 @@ class TestDoctests:
assert isinstance(items[0].parent, DoctestModule)
assert items[0].parent is items[1].parent
@pytest.mark.xfail('hasattr(sys, "pypy_version_info")', reason=
"pypy leaks one FD")
def test_simple_doctestfile(self, testdir):
p = testdir.maketxtfile(test_doc="""
>>> x = 1

View File

@ -16,7 +16,6 @@ class Standalone:
assert self.script.check()
def run(self, anypython, testdir, *args):
testdir.chdir()
return testdir._run(anypython, self.script, *args)
def test_gen(testdir, anypython, standalone):

View File

@ -53,14 +53,14 @@ def test_traceconfig(testdir):
])
def test_debug(testdir, monkeypatch):
result = testdir.runpytest("--debug")
result = testdir.runpytest_subprocess("--debug")
assert result.ret == 0
p = testdir.tmpdir.join("pytestdebug.log")
assert "pytest_sessionstart" in p.read()
def test_PYTEST_DEBUG(testdir, monkeypatch):
monkeypatch.setenv("PYTEST_DEBUG", "1")
result = testdir.runpytest()
result = testdir.runpytest_subprocess()
assert result.ret == 0
result.stderr.fnmatch_lines([
"*pytest_plugin_registered*",

View File

@ -260,7 +260,7 @@ class TestPDB:
def test_pdb_collection_failure_is_shown(self, testdir):
p1 = testdir.makepyfile("""xxx """)
result = testdir.runpytest("--pdb", p1)
result = testdir.runpytest_subprocess("--pdb", p1)
result.stdout.fnmatch_lines([
"*NameError*xxx*",
"*1 error*",

View File

@ -203,7 +203,6 @@ class TestNewSession(SessionTests):
def test_plugin_specify(testdir):
testdir.chdir()
pytest.raises(ImportError, """
testdir.parseconfig("-p", "nqweotexistent")
""")