Parallel scan test fails
@valentin.valls Parallel scan test fails sometimes:
_____________________________ test_parallel_scans ______________________________
default_session = <bliss.common.session.DefaultSession object at 0x7fad453e18d0>
scan_tmpdir = local('/tmp/pytest-of-root/pytest-0/test_parallel_scans0')
def test_parallel_scans(default_session, scan_tmpdir):
default_session.scan_saving.base_path = str(scan_tmpdir)
diode = default_session.config.get("diode")
sim_ct_gauss = default_session.config.get("sim_ct_gauss")
robz = default_session.config.get("robz")
s1 = scans.loopscan(20, .1, diode, run=False)
s2 = scans.ascan(robz, 0, 10, 25, .09, sim_ct_gauss, run=False)
g1 = gevent.spawn(s1.run)
g2 = gevent.spawn(s2.run)
gs = [g1, g2]
new_scan_args = []
new_child_args = []
new_data_args = []
end_scan_args = []
end_scan_event = gevent.event.Event()
def end(*args):
end_scan_event.set()
end_scan_args.append(args)
session_watcher = gevent.spawn(
watch_session_scans,
default_session.name,
lambda *args: new_scan_args.append(args),
lambda *args: new_child_args.append(args),
lambda *args: new_data_args.append(args),
end,
)
try:
gevent.sleep(0.1) # wait a bit to have session watcher greenlet started
gevent.joinall(gs, raise_error=True)
end_scan_event.wait(2.0)
finally:
session_watcher.kill()
> assert len(new_data_args) > 0
E assert 0 > 0
E + where 0 = len([])