637 lines
20 KiB
Python
637 lines
20 KiB
Python
import py
|
|
import pytest
|
|
|
|
|
|
class TestDistribution:
|
|
def test_n1_pass(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
def test_ok():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest(p1, "-n1")
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines(["*1 passed*", ])
|
|
|
|
def test_n1_fail(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
def test_fail():
|
|
assert 0
|
|
""")
|
|
result = testdir.runpytest(p1, "-n1")
|
|
assert result.ret == 1
|
|
result.stdout.fnmatch_lines(["*1 failed*", ])
|
|
|
|
def test_n1_import_error(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
import __import_of_missing_module
|
|
def test_import():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest(p1, "-n1")
|
|
assert result.ret == 1
|
|
result.stdout.fnmatch_lines([
|
|
"E ImportError: *__import_of_missing_module*",
|
|
])
|
|
|
|
def test_n2_import_error(self, testdir):
|
|
"""Check that we don't report the same import error multiple times
|
|
in distributed mode."""
|
|
p1 = testdir.makepyfile("""
|
|
import __import_of_missing_module
|
|
def test_import():
|
|
pass
|
|
""")
|
|
result1 = testdir.runpytest(p1, "-n2")
|
|
result2 = testdir.runpytest(p1, "-n1")
|
|
assert len(result1.stdout.lines) == len(result2.stdout.lines)
|
|
|
|
def test_n1_skip(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
def test_skip():
|
|
import py
|
|
py.test.skip("myreason")
|
|
""")
|
|
result = testdir.runpytest(p1, "-n1")
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines(["*1 skipped*", ])
|
|
|
|
def test_manytests_to_one_import_error(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
import __import_of_missing_module
|
|
def test_import():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest(p1, '--tx=popen', '--tx=popen')
|
|
assert result.ret == 1
|
|
result.stdout.fnmatch_lines([
|
|
"E ImportError: *__import_of_missing_module*",
|
|
])
|
|
|
|
def test_manytests_to_one_popen(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
import py
|
|
def test_fail0():
|
|
assert 0
|
|
def test_fail1():
|
|
raise ValueError()
|
|
def test_ok():
|
|
pass
|
|
def test_skip():
|
|
py.test.skip("hello")
|
|
""", )
|
|
result = testdir.runpytest(p1, "-v", '-d', '--tx=popen', '--tx=popen')
|
|
result.stdout.fnmatch_lines([
|
|
"*1*Python*",
|
|
"*2 failed, 1 passed, 1 skipped*",
|
|
])
|
|
assert result.ret == 1
|
|
|
|
def test_n1_fail_minus_x(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
def test_fail1():
|
|
assert 0
|
|
def test_fail2():
|
|
assert 0
|
|
""")
|
|
result = testdir.runpytest(p1, "-x", "-v", "-n1")
|
|
assert result.ret == 2
|
|
result.stdout.fnmatch_lines([
|
|
"*Interrupted: stopping*1*",
|
|
"*1 failed*",
|
|
])
|
|
|
|
def test_basetemp_in_subprocesses(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
def test_send(tmpdir):
|
|
import py
|
|
assert tmpdir.relto(py.path.local(%r)), tmpdir
|
|
""" % str(testdir.tmpdir))
|
|
result = testdir.runpytest_subprocess(p1, "-n1")
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines(["*1 passed*", ])
|
|
|
|
def test_dist_ini_specified(self, testdir):
|
|
p1 = testdir.makepyfile("""
|
|
import py
|
|
def test_fail0():
|
|
assert 0
|
|
def test_fail1():
|
|
raise ValueError()
|
|
def test_ok():
|
|
pass
|
|
def test_skip():
|
|
py.test.skip("hello")
|
|
""", )
|
|
testdir.makeini("""
|
|
[pytest]
|
|
addopts = --tx=3*popen
|
|
""")
|
|
result = testdir.runpytest(p1, '-d', "-v")
|
|
result.stdout.fnmatch_lines([
|
|
"*2*Python*",
|
|
"*2 failed, 1 passed, 1 skipped*",
|
|
])
|
|
assert result.ret == 1
|
|
|
|
@py.test.mark.xfail("sys.platform.startswith('java')", run=False)
|
|
def test_dist_tests_with_crash(self, testdir):
|
|
if not hasattr(py.std.os, 'kill'):
|
|
py.test.skip("no os.kill")
|
|
|
|
p1 = testdir.makepyfile("""
|
|
import py
|
|
def test_fail0():
|
|
assert 0
|
|
def test_fail1():
|
|
raise ValueError()
|
|
def test_ok():
|
|
pass
|
|
def test_skip():
|
|
py.test.skip("hello")
|
|
def test_crash():
|
|
import time
|
|
import os
|
|
time.sleep(0.5)
|
|
os.kill(os.getpid(), 15)
|
|
""")
|
|
result = testdir.runpytest(p1, "-v", '-d', '-n1')
|
|
result.stdout.fnmatch_lines([
|
|
"*Python*", "*PASS**test_ok*", "*node*down*",
|
|
"*3 failed, 1 passed, 1 skipped*"
|
|
])
|
|
assert result.ret == 1
|
|
|
|
def test_distribution_rsyncdirs_example(self, testdir):
|
|
source = testdir.mkdir("source")
|
|
dest = testdir.mkdir("dest")
|
|
subdir = source.mkdir("example_pkg")
|
|
subdir.ensure("__init__.py")
|
|
p = subdir.join("test_one.py")
|
|
p.write("def test_5():\n assert not __file__.startswith(%r)" % str(p))
|
|
result = testdir.runpytest("-v", "-d",
|
|
"--rsyncdir=%(subdir)s" % locals(),
|
|
"--tx=popen//chdir=%(dest)s" % locals(), p)
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines([
|
|
"*0* *cwd*",
|
|
# "RSyncStart: [G1]",
|
|
# "RSyncFinished: [G1]",
|
|
"*1 passed*"
|
|
])
|
|
assert dest.join(subdir.basename).check(dir=1)
|
|
|
|
def test_data_exchange(self, testdir):
|
|
testdir.makeconftest("""
|
|
# This hook only called on master.
|
|
def pytest_configure_node(node):
|
|
node.slaveinput['a'] = 42
|
|
node.slaveinput['b'] = 7
|
|
|
|
def pytest_configure(config):
|
|
# this attribute is only set on slaves
|
|
if hasattr(config, 'slaveinput'):
|
|
a = config.slaveinput['a']
|
|
b = config.slaveinput['b']
|
|
r = a + b
|
|
config.slaveoutput['r'] = r
|
|
|
|
# This hook only called on master.
|
|
def pytest_testnodedown(node, error):
|
|
node.config.calc_result = node.slaveoutput['r']
|
|
|
|
def pytest_terminal_summary(terminalreporter):
|
|
if not hasattr(terminalreporter.config, 'slaveinput'):
|
|
calc_result = terminalreporter.config.calc_result
|
|
terminalreporter._tw.sep('-',
|
|
'calculated result is %s' % calc_result)
|
|
""")
|
|
p1 = testdir.makepyfile("def test_func(): pass")
|
|
result = testdir.runpytest("-v", p1, '-d', '--tx=popen')
|
|
result.stdout.fnmatch_lines([
|
|
"*0*Python*", "*calculated result is 49*", "*1 passed*"
|
|
])
|
|
assert result.ret == 0
|
|
|
|
def test_keyboardinterrupt_hooks_issue79(self, testdir):
|
|
testdir.makepyfile(__init__="",
|
|
test_one="""
|
|
def test_hello():
|
|
raise KeyboardInterrupt()
|
|
""")
|
|
testdir.makeconftest("""
|
|
def pytest_sessionfinish(session):
|
|
# on the slave
|
|
if hasattr(session.config, 'slaveoutput'):
|
|
session.config.slaveoutput['s2'] = 42
|
|
# on the master
|
|
def pytest_testnodedown(node, error):
|
|
assert node.slaveoutput['s2'] == 42
|
|
print ("s2call-finished")
|
|
""")
|
|
args = ["-n1", "--debug"]
|
|
result = testdir.runpytest_subprocess(*args)
|
|
s = result.stdout.str()
|
|
assert result.ret == 2
|
|
assert 's2call' in s
|
|
assert "Interrupted" in s
|
|
|
|
def test_keyboard_interrupt_dist(self, testdir):
|
|
# xxx could be refined to check for return code
|
|
testdir.makepyfile("""
|
|
def test_sleep():
|
|
import time
|
|
time.sleep(10)
|
|
""")
|
|
child = testdir.spawn_pytest("-n1 -v")
|
|
child.expect(".*test_sleep.*")
|
|
child.kill(2) # keyboard interrupt
|
|
child.expect(".*KeyboardInterrupt.*")
|
|
# child.expect(".*seconds.*")
|
|
child.close()
|
|
# assert ret == 2
|
|
|
|
|
|
class TestDistEach:
|
|
def test_simple(self, testdir):
|
|
testdir.makepyfile("""
|
|
def test_hello():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest_subprocess("--debug", "--dist=each",
|
|
"--tx=2*popen")
|
|
assert not result.ret
|
|
result.stdout.fnmatch_lines(["*2 pass*"])
|
|
|
|
@py.test.mark.xfail(
|
|
run=False,
|
|
reason="other python versions might not have py.test installed")
|
|
def test_simple_diffoutput(self, testdir):
|
|
interpreters = []
|
|
for name in ("python2.5", "python2.6"):
|
|
interp = py.path.local.sysfind(name)
|
|
if interp is None:
|
|
py.test.skip("%s not found" % name)
|
|
interpreters.append(interp)
|
|
|
|
testdir.makepyfile(__init__="",
|
|
test_one="""
|
|
import sys
|
|
def test_hello():
|
|
print("%s...%s" % sys.version_info[:2])
|
|
assert 0
|
|
""")
|
|
args = ["--dist=each", "-v"]
|
|
args += ["--tx", "popen//python=%s" % interpreters[0]]
|
|
args += ["--tx", "popen//python=%s" % interpreters[1]]
|
|
result = testdir.runpytest(*args)
|
|
s = result.stdout.str()
|
|
assert "2...5" in s
|
|
assert "2...6" in s
|
|
|
|
|
|
class TestTerminalReporting:
|
|
def test_pass_skip_fail(self, testdir):
|
|
testdir.makepyfile("""
|
|
import py
|
|
def test_ok():
|
|
pass
|
|
def test_skip():
|
|
py.test.skip("xx")
|
|
def test_func():
|
|
assert 0
|
|
""")
|
|
result = testdir.runpytest("-n1", "-v")
|
|
result.stdout.fnmatch_lines_random([
|
|
"*PASS*test_pass_skip_fail.py*test_ok*",
|
|
"*SKIP*test_pass_skip_fail.py*test_skip*",
|
|
"*FAIL*test_pass_skip_fail.py*test_func*",
|
|
])
|
|
result.stdout.fnmatch_lines([
|
|
"*def test_func():",
|
|
"> assert 0",
|
|
"E assert 0",
|
|
])
|
|
|
|
def test_fail_platinfo(self, testdir):
|
|
testdir.makepyfile("""
|
|
def test_func():
|
|
assert 0
|
|
""")
|
|
result = testdir.runpytest("-n1", "-v")
|
|
result.stdout.fnmatch_lines([
|
|
"*FAIL*test_fail_platinfo.py*test_func*",
|
|
"*0*Python*",
|
|
"*def test_func():",
|
|
"> assert 0",
|
|
"E assert 0",
|
|
])
|
|
|
|
|
|
def test_teardownfails_one_function(testdir):
|
|
p = testdir.makepyfile("""
|
|
def test_func():
|
|
pass
|
|
def teardown_function(function):
|
|
assert 0
|
|
""")
|
|
result = testdir.runpytest(p, '-n1', '--tx=popen')
|
|
result.stdout.fnmatch_lines([
|
|
"*def teardown_function(function):*", "*1 passed*1 error*"
|
|
])
|
|
|
|
|
|
@py.test.mark.xfail
|
|
def test_terminate_on_hangingnode(testdir):
|
|
p = testdir.makeconftest("""
|
|
def pytest_sessionfinish(session):
|
|
if session.nodeid == "my": # running on slave
|
|
import time
|
|
time.sleep(3)
|
|
""")
|
|
result = testdir.runpytest(p, '--dist=each', '--tx=popen//id=my')
|
|
assert result.duration < 2.0
|
|
result.stdout.fnmatch_lines(["*killed*my*", ])
|
|
|
|
|
|
@pytest.mark.xfail(reason='see #30')
|
|
def test_auto_detect_cpus(testdir, monkeypatch):
|
|
import multiprocessing
|
|
monkeypatch.setattr(multiprocessing, 'cpu_count', lambda: 3)
|
|
testdir.makeconftest("""
|
|
def pytest_unconfigure(config):
|
|
with open('cpus', 'w') as f:
|
|
f.write('cpus = %s' % config.option.numprocesses)
|
|
""")
|
|
testdir.inline_run('-n=auto')
|
|
cpus_file = testdir.tmpdir.join('cpus')
|
|
assert cpus_file.read() == 'cpus = 3'
|
|
|
|
|
|
@pytest.mark.xfail(reason="works if run outside test suite", run=False)
|
|
def test_session_hooks(testdir):
|
|
testdir.makeconftest("""
|
|
import sys
|
|
def pytest_sessionstart(session):
|
|
sys.pytestsessionhooks = session
|
|
def pytest_sessionfinish(session):
|
|
if hasattr(session.config, 'slaveinput'):
|
|
name = "slave"
|
|
else:
|
|
name = "master"
|
|
f = open(name, "w")
|
|
f.write("xy")
|
|
f.close()
|
|
# let's fail on the slave
|
|
if name == "slave":
|
|
raise ValueError(42)
|
|
""")
|
|
p = testdir.makepyfile("""
|
|
import sys
|
|
def test_hello():
|
|
assert hasattr(sys, 'pytestsessionhooks')
|
|
""")
|
|
result = testdir.runpytest(p, "--dist=each", "--tx=popen")
|
|
result.stdout.fnmatch_lines(["*ValueError*", "*1 passed*", ])
|
|
assert not result.ret
|
|
d = result.parseoutcomes()
|
|
assert d['passed'] == 1
|
|
assert testdir.tmpdir.join("slave").check()
|
|
assert testdir.tmpdir.join("master").check()
|
|
|
|
|
|
def test_session_testscollected(testdir):
|
|
"""
|
|
Make sure master node is updating the session object with the number
|
|
of tests collected from the slaves.
|
|
"""
|
|
testdir.makepyfile(test_foo="""
|
|
import pytest
|
|
@pytest.mark.parametrize('i', range(3))
|
|
def test_ok(i):
|
|
pass
|
|
""")
|
|
testdir.makeconftest("""
|
|
def pytest_sessionfinish(session):
|
|
collected = getattr(session, 'testscollected', None)
|
|
with open('testscollected', 'w') as f:
|
|
f.write('collected = %s' % collected)
|
|
""")
|
|
result = testdir.inline_run("-n1")
|
|
result.assertoutcome(passed=3)
|
|
collected_file = testdir.tmpdir.join('testscollected')
|
|
assert collected_file.isfile()
|
|
assert collected_file.read() == 'collected = 3'
|
|
|
|
|
|
def test_funcarg_teardown_failure(testdir):
|
|
p = testdir.makepyfile("""
|
|
def pytest_funcarg__myarg(request):
|
|
def teardown(val):
|
|
raise ValueError(val)
|
|
return request.cached_setup(setup=lambda: 42, teardown=teardown,
|
|
scope="module")
|
|
def test_hello(myarg):
|
|
pass
|
|
""")
|
|
result = testdir.runpytest_subprocess("--debug", p) # , "-n1")
|
|
result.stdout.fnmatch_lines(["*ValueError*42*", "*1 passed*1 error*", ])
|
|
assert result.ret
|
|
|
|
|
|
def test_crashing_item(testdir):
|
|
p = testdir.makepyfile("""
|
|
import py
|
|
import os
|
|
def test_crash():
|
|
py.process.kill(os.getpid())
|
|
def test_noncrash():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest("-n2", p)
|
|
result.stdout.fnmatch_lines([
|
|
"*crashed*test_crash*", "*1 failed*1 passed*"
|
|
])
|
|
|
|
|
|
def test_skipping(testdir):
|
|
p = testdir.makepyfile("""
|
|
import pytest
|
|
def test_crash():
|
|
pytest.skip("hello")
|
|
""")
|
|
result = testdir.runpytest("-n1", '-rs', p)
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines(["*hello*", "*1 skipped*"])
|
|
|
|
|
|
def test_issue34_pluginloading_in_subprocess(testdir):
|
|
testdir.tmpdir.join("plugin123.py").write(py.code.Source("""
|
|
def pytest_namespace():
|
|
return {'sample_variable': 'testing'}
|
|
"""))
|
|
testdir.makepyfile("""
|
|
import pytest
|
|
def test_hello():
|
|
assert pytest.sample_variable == "testing"
|
|
""")
|
|
result = testdir.runpytest_subprocess("-n1", "-p", "plugin123")
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines(["*1 passed*", ])
|
|
|
|
|
|
def test_fixture_scope_caching_issue503(testdir):
|
|
p1 = testdir.makepyfile("""
|
|
import pytest
|
|
|
|
@pytest.fixture(scope='session')
|
|
def fix():
|
|
assert fix.counter == 0, \
|
|
'session fixture was invoked multiple times'
|
|
fix.counter += 1
|
|
fix.counter = 0
|
|
|
|
def test_a(fix):
|
|
pass
|
|
|
|
def test_b(fix):
|
|
pass
|
|
""")
|
|
result = testdir.runpytest(p1, '-v', '-n1')
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines(["*2 passed*", ])
|
|
|
|
|
|
def test_issue_594_random_parametrize(testdir):
|
|
"""
|
|
Make sure that tests that are randomly parametrized display an appropriate
|
|
error message, instead of silently skipping the entire test run.
|
|
"""
|
|
p1 = testdir.makepyfile("""
|
|
import pytest
|
|
import random
|
|
|
|
xs = list(range(10))
|
|
random.shuffle(xs)
|
|
@pytest.mark.parametrize('x', xs)
|
|
def test_foo(x):
|
|
assert 1
|
|
""")
|
|
result = testdir.runpytest(p1, '-v', '-n4')
|
|
assert result.ret == 1
|
|
result.stdout.fnmatch_lines([
|
|
"Different tests were collected between gw* and gw*",
|
|
])
|
|
|
|
|
|
def test_tmpdir_disabled(testdir):
|
|
"""Test xdist doesn't break if internal tmpdir plugin is disabled (#22).
|
|
"""
|
|
p1 = testdir.makepyfile("""
|
|
def test_ok():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest(p1, "-n1", '-p', 'no:tmpdir')
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines("*1 passed*")
|
|
|
|
|
|
@pytest.mark.parametrize('plugin', ['xdist.looponfail', 'xdist.boxed'])
|
|
def test_sub_plugins_disabled(testdir, plugin):
|
|
"""Test that xdist doesn't break if we disable any of its sub-plugins. (#32)
|
|
"""
|
|
p1 = testdir.makepyfile("""
|
|
def test_ok():
|
|
pass
|
|
""")
|
|
result = testdir.runpytest(p1, "-n1", '-p', 'no:%s' % plugin)
|
|
assert result.ret == 0
|
|
result.stdout.fnmatch_lines("*1 passed*")
|
|
|
|
|
|
class TestNodeFailure:
|
|
def test_load_single(self, testdir):
|
|
f = testdir.makepyfile("""
|
|
import os
|
|
def test_a(): os._exit(1)
|
|
def test_b(): pass
|
|
""")
|
|
res = testdir.runpytest(f, '-n1')
|
|
res.stdout.fnmatch_lines([
|
|
"*Replacing crashed slave*",
|
|
"*Slave*crashed while running*",
|
|
"*1 failed*1 passed*",
|
|
])
|
|
|
|
def test_load_multiple(self, testdir):
|
|
f = testdir.makepyfile("""
|
|
import os
|
|
def test_a(): pass
|
|
def test_b(): os._exit(1)
|
|
def test_c(): pass
|
|
def test_d(): pass
|
|
""")
|
|
res = testdir.runpytest(f, '-n2')
|
|
res.stdout.fnmatch_lines([
|
|
"*Replacing crashed slave*",
|
|
"*Slave*crashed while running*",
|
|
"*1 failed*3 passed*",
|
|
])
|
|
|
|
def test_each_single(self, testdir):
|
|
f = testdir.makepyfile("""
|
|
import os
|
|
def test_a(): os._exit(1)
|
|
def test_b(): pass
|
|
""")
|
|
res = testdir.runpytest(f, '--dist=each', '--tx=popen')
|
|
res.stdout.fnmatch_lines([
|
|
"*Replacing crashed slave*",
|
|
"*Slave*crashed while running*",
|
|
"*1 failed*1 passed*",
|
|
])
|
|
|
|
@pytest.mark.xfail(reason='#20: xdist race condition on node restart')
|
|
def test_each_multiple(self, testdir):
|
|
f = testdir.makepyfile("""
|
|
import os
|
|
def test_a(): os._exit(1)
|
|
def test_b(): pass
|
|
""")
|
|
res = testdir.runpytest(f, '--dist=each', '--tx=2*popen')
|
|
res.stdout.fnmatch_lines([
|
|
"*Replacing crashed slave*",
|
|
"*Slave*crashed while running*",
|
|
"*2 failed*2 passed*",
|
|
])
|
|
|
|
def test_max_slave_restart(self, testdir):
|
|
f = testdir.makepyfile("""
|
|
import os
|
|
def test_a(): pass
|
|
def test_b(): os._exit(1)
|
|
def test_c(): os._exit(1)
|
|
def test_d(): pass
|
|
""")
|
|
res = testdir.runpytest(f, '-n4', '--max-slave-restart=1')
|
|
res.stdout.fnmatch_lines([
|
|
"*Replacing crashed slave*",
|
|
"*Maximum crashed slaves reached: 1*",
|
|
"*Slave*crashed while running*",
|
|
"*Slave*crashed while running*",
|
|
"*2 failed*2 passed*",
|
|
])
|
|
|
|
def test_disable_restart(self, testdir):
|
|
f = testdir.makepyfile("""
|
|
import os
|
|
def test_a(): pass
|
|
def test_b(): os._exit(1)
|
|
def test_c(): pass
|
|
""")
|
|
res = testdir.runpytest(f, '-n4', '--max-slave-restart=0')
|
|
res.stdout.fnmatch_lines([
|
|
"*Slave restarting disabled*",
|
|
"*Slave*crashed while running*",
|
|
"*1 failed*2 passed*",
|
|
])
|