Commit 8f59b689 authored by Kirill Smelkov's avatar Kirill Smelkov

tests: Run nxdtest.main for each test in a separate thread, so that pytest.timeout generally works

Factor-out the driver to run_nxdtest in a separate thread done in
0ad45a9c (Detect if a test leaks processes and terminate them) from
test_run_procleak into run_nxdtest, so that all tests are run this way
for pytest.timeout, if requested, to work universally for all tests.

The logic of how we run nxdtest under tests and handle timeout will be,
hopefully, soon reworked more, but it is anyway good to keep this logic
in only one place.

/reviewed-by @jerome
/reviewed-on !14
parent 938b5455
...@@ -38,16 +38,38 @@ def run_nxdtest(tmpdir): ...@@ -38,16 +38,38 @@ def run_nxdtest(tmpdir):
passed as `argv`. passed as `argv`.
""" """
@func
def _run_nxdtest(nxdtest_file_content, argv=("nxdtest",)): def _run_nxdtest(nxdtest_file_content, argv=("nxdtest",)):
with tmpdir.as_cwd(): with tmpdir.as_cwd():
with open(".nxdtest", "w") as f: with open(".nxdtest", "w") as f:
f.write(nxdtest_file_content) f.write(nxdtest_file_content)
sys_argv = sys.argv sys_argv = sys.argv
sys.argv = argv sys.argv = argv
try: def _():
main()
finally:
sys.argv = sys_argv sys.argv = sys_argv
defer(_)
# run nxdtest in thread so that timeout handling works
# ( if nxdtest is run on main thread, then non-py wait in WorkGroup.wait, if
# stuck, prevents signals from being handled at python-level )
wg = sync.WorkGroup(context.background())
done = chan()
@func
def _(ctx):
defer(done.close)
main()
wg.go(_)
while 1:
_, _rx = select(
default, # 0
done.recv, # 1
)
if _ == 0:
time.sleep(0.1)
continue
wg.wait()
break
return _run_nxdtest return _run_nxdtest
...@@ -134,32 +156,11 @@ TestCase('TEST10', ['echo', 'TEST10']) ...@@ -134,32 +156,11 @@ TestCase('TEST10', ['echo', 'TEST10'])
def test_run_procleak(run_nxdtest, capsys): def test_run_procleak(run_nxdtest, capsys):
procleak = "%s/testprog/procleak" % (dirname(__file__),) procleak = "%s/testprog/procleak" % (dirname(__file__),)
# run nxdtest in thread so that timeout handling works
# ( if nxdtest is run on main thread, then non-py wait in WorkGroup.wait, if
# stuck, prevents signals from being handled at python-level )
wg = sync.WorkGroup(context.background())
done = chan()
@func
def _(ctx):
defer(done.close)
run_nxdtest( run_nxdtest(
"""\ """\
TestCase('TEST_WITH_PROCLEAK', ['%s', 'AAA', 'BBB', 'CCC']) TestCase('TEST_WITH_PROCLEAK', ['%s', 'AAA', 'BBB', 'CCC'])
""" % procleak """ % procleak
) )
wg.go(_)
while 1:
_, _rx = select(
default, # 0
done.recv, # 1
)
if _ == 0:
time.sleep(0.1)
continue
wg.wait()
break
captured = capsys.readouterr() captured = capsys.readouterr()
assert "AAA: terminating" in captured.out assert "AAA: terminating" in captured.out
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment