def test_looponfailing(self): |
o = tmpdir.ensure('looponfailing', dir=1) |
tfile = o.join('test_looponfailing.py') |
tfile.write(py.code.Source(""" |
def test_1(): |
assert 1 == 0 |
""")) |
print py.std.sys.executable |
config = py.test.config._reparse(['--looponfailing', str(o)]) |
cls = config._getsessionclass() |
out = py.std.Queue.Queue() |
session = cls(config, out.put) |
-> pool = py._thread.WorkerPool() |
reply = pool.dispatch(session.main) |
while 1: |
s = out.get(timeout=5.0) |
if s.find('1 failed') != -1: |
break |
print s |
else: |
py.test.fail("did not see test_1 failure") |
|
try: |
reply.get(timeout=5.0) |
except IOError, e: |
assert str(e).lower().find('timeout') != -1 |