IQ.Lvbs/pytest-xdist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move some code around, ending up with fewer files logically grouped

Browse Source
This commit is contained in:
holger krekel
2010-09-28 16:08:31 +02:00
parent f7923c692e
commit 3767ce3269
11 changed files with 236 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
testing/acceptance_test.py
View File

@@ -272,3 +272,101 @@ class TestTerminalReporting:
"> 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__teardown_final(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*",
])
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_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("--debug", p, "-n1")
result.stdout.fnmatch_lines([
"*ValueError*42*",
"*1 passed*1 error*",
])
py.test.xfail("fix exitstatus handling")
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*"
])

112
testing/test_dsession.py
View File

@@ -429,115 +429,3 @@ class TestDSession:
linecomp.assert_contains_lines([
"[X1,X2] rsyncing: hello",
])
def test_collected_function_causes_remote_skip(testdir):
sub = testdir.mkpydir("testing")
sub.join("test_module.py").write(py.code.Source("""
import py
path = py.path.local(%r)
if path.check():
path.remove()
else:
py.test.skip("remote skip")
def test_func():
pass
def test_func2():
pass
""" % str(sub.ensure("somefile"))))
result = testdir.runpytest('-v', '--dist=each', '--tx=popen')
result.stdout.fnmatch_lines([
"*2 skipped*"
])
def test_teardownfails_one_function(testdir):
p = testdir.makepyfile("""
def test_func():
pass
def teardown_function(function):
assert 0
""")
result = testdir.runpytest(p, '--dist=each', '--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__teardown_final(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*",
])
def test_session_hooks(testdir):
testdir.makeconftest("""
import sys
def pytest_sessionstart(session):
sys.pytestsessionhooks = session
def pytest_sessionfinish(session):
f = open(session.nodeid or "master", 'w')
f.write("xy")
f.close()
# let's fail on the slave
if session.nodeid:
raise ValueError(42)
""")
p = testdir.makepyfile("""
import sys
def test_hello():
assert hasattr(sys, 'pytestsessionhooks')
""")
result = testdir.runpytest(p, "--dist=each", "--tx=popen//id=my1")
result.stdout.fnmatch_lines([
"*ValueError*",
"*1 passed*",
])
assert result.ret
d = result.parseoutcomes()
assert d['passed'] == 1
assert testdir.tmpdir.join("my1").check()
assert testdir.tmpdir.join("master").check()
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(p, "-n1")
assert result.ret
result.stdout.fnmatch_lines([
"*ValueError*42*",
"*1 passed*1 error*",
])
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*"
])

61
testing/test_looponfail.py
View File

@@ -1,5 +1,65 @@
import py
from xdist.looponfail import RemoteControl
from xdist.looponfail import StatRecorder
class TestStatRecorder:
def test_filechange(self, tmpdir):
tmp = tmpdir
hello = tmp.ensure("hello.py")
sd = StatRecorder([tmp])
changed = sd.check()
assert not changed
hello.write("world")
changed = sd.check()
assert changed
tmp.ensure("new.py")
changed = sd.check()
assert changed
tmp.join("new.py").remove()
changed = sd.check()
assert changed
tmp.join("a", "b", "c.py").ensure()
changed = sd.check()
assert changed
tmp.join("a", "c.txt").ensure()
changed = sd.check()
assert changed
changed = sd.check()
assert not changed
tmp.join("a").remove()
changed = sd.check()
assert changed
def test_pycremoval(self, tmpdir):
tmp = tmpdir
hello = tmp.ensure("hello.py")
sd = StatRecorder([tmp])
changed = sd.check()
assert not changed
pycfile = hello + "c"
pycfile.ensure()
changed = sd.check()
assert not changed
hello.write("world")
changed = sd.check()
assert not pycfile.check()
def test_waitonchange(self, tmpdir, monkeypatch):
tmp = tmpdir
sd = StatRecorder([tmp])
l = [True, False]
monkeypatch.setattr(StatRecorder, 'check', lambda self: l.pop())
sd.waitonchange(checkinterval=0.2)
assert not l
class TestRemoteControl:
def test_nofailures(self, testdir):
@@ -174,3 +234,4 @@ def removepyc(path):
pyc = path + "c"
if pyc.check():
pyc.remove()

12
testing/test_remote.py
View File

@@ -83,7 +83,7 @@ class TestReportSerialization:
for rep in reports:
d = serialize_report(rep)
check_marshallable(d)
newrep = unserialize_report(d)
newrep = unserialize_report("testreport", d)
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
@@ -99,7 +99,7 @@ class TestReportSerialization:
for rep in reports:
d = serialize_report(rep)
check_marshallable(d)
newrep = unserialize_report(d)
newrep = unserialize_report("collectreport", d)
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
@@ -111,7 +111,7 @@ class TestReportSerialization:
for rep in reports:
d = serialize_report(rep)
check_marshallable(d)
newrep = unserialize_report(d)
newrep = unserialize_report("collectreport", d)
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
@@ -137,7 +137,7 @@ class TestSlaveInteractor:
slave.sendcommand("runtests", ids=ids)
ev = slave.popevent("testreport")
assert ev.name == "testreport"
rep = unserialize_report(ev.kwargs['data'])
rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.nodeid.endswith("::test_func")
assert rep.passed
assert rep.when == "call"
@@ -155,7 +155,7 @@ class TestSlaveInteractor:
assert not ev.kwargs
ev = slave.popevent()
assert ev.name == "collectreport"
rep = unserialize_report(ev.kwargs['data'])
rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.skipped
ev = slave.popevent("collectionfinish")
print ev.kwargs
@@ -168,7 +168,7 @@ class TestSlaveInteractor:
assert not ev.kwargs
ev = slave.popevent()
assert ev.name == "collectreport"
rep = unserialize_report(ev.kwargs['data'])
rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.failed
ev = slave.popevent("collectionfinish")
print ev.kwargs

61
testing/test_util.py
View File

@@ -1,61 +0,0 @@
import py
from xdist.util import StatRecorder
def test_filechange(tmpdir):
tmp = tmpdir
hello = tmp.ensure("hello.py")
sd = StatRecorder([tmp])
changed = sd.check()
assert not changed
hello.write("world")
changed = sd.check()
assert changed
tmp.ensure("new.py")
changed = sd.check()
assert changed
tmp.join("new.py").remove()
changed = sd.check()
assert changed
tmp.join("a", "b", "c.py").ensure()
changed = sd.check()
assert changed
tmp.join("a", "c.txt").ensure()
changed = sd.check()
assert changed
changed = sd.check()
assert not changed
tmp.join("a").remove()
changed = sd.check()
assert changed
def test_pycremoval(tmpdir):
tmp = tmpdir
hello = tmp.ensure("hello.py")
sd = StatRecorder([tmp])
changed = sd.check()
assert not changed
pycfile = hello + "c"
pycfile.ensure()
changed = sd.check()
assert not changed
hello.write("world")
changed = sd.check()
assert not pycfile.check()
def test_waitonchange(tmpdir, monkeypatch):
tmp = tmpdir
sd = StatRecorder([tmp])
l = [True, False]
monkeypatch.setattr(StatRecorder, 'check', lambda self: l.pop())
sd.waitonchange(checkinterval=0.2)
assert not l