CI: test_nxw_readers_append flaky
https://gitlab.esrf.fr/bliss/bliss/-/jobs/107620
___________________________ test_nxw_readers_append ____________________________
nexus_writer_config = {'alt': False, 'config': True, 'policy': True, 'session': <bliss.common.session.Session object at 0x7f161877ddd0>, ...}
def test_nxw_readers_append(nexus_writer_config):
> _test_nxw_readers(mode="a", enable_file_locking=False, **nexus_writer_config)
tests/nexus_writer/test_nxw_readers.py:27:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/nexus_writer/helpers/nxw_test_utils.py:291: in inner
func(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
mode = 'a', enable_file_locking = False
session = <bliss.common.session.Session object at 0x7f161877ddd0>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_nxw_readers_append0')
writer = <PopenGreenlet at 0x7f16183e3ef0: _run>, config = True
kwargs = {'alt': False, 'policy': True}, nxroot = <Closed HDF5 file>
nxentry = <Closed HDF5 group>, measurement = <Closed HDF5 group>, readers = []
detector = 'diode3', scan_shape = (100,)
scan = Scan(number=1, name=loopscan, path=/tmp/pytest-of-root/pytest-0/test_nxw_readers_append0/nexus_writer_session/tmp/testproposal/id00/sample/sample_0001/sample_0001.h5)
@nxw_test_utils.writer_stdout_on_exception
def _test_nxw_readers(
mode="r",
enable_file_locking=False,
session=None,
tmpdir=None,
writer=None,
config=True,
**kwargs
):
session.scan_saving.technique = "none"
filename = scan_utils.session_filename(scan_saving=session.scan_saving)
with nexus.nxRoot(filename, mode="a") as nxroot:
nxentry = nexus.nxEntry(nxroot, "dummy")
measurement = nexus.nxCollection(nxentry, "measurement")
measurement["a"] = list(range(10))
measurement["b"] = list(range(10))
startevent = gevent.event.Event()
readerkwargs = {"mode": mode, "enable_file_locking": enable_file_locking}
readers = [
gevent.spawn(reader, filename, startevent, hold=i == 0, **readerkwargs)
for i in range(4)
]
startevent.wait()
try:
detector = "diode3"
if mode == "a" and not enable_file_locking:
# Readers will not crash the scan (not sure why) but corrupt the file
scan_shape = (100,)
scan = scans.loopscan(
scan_shape[0], .1, session.env_dict[detector], run=False
)
nxw_test_utils.run_scan(scan)
gevent.killall(readers)
gevent.joinall(readers)
readers = []
with pytest.raises(AssertionError):
# We can only detect the corruption after closing the readers
> nxw_test_utils.assert_scan_data_not_corrupt([scan])
E Failed: DID NOT RAISE <class 'AssertionError'>
tests/nexus_writer/test_nxw_readers.py:72: Failed
Edited by Wout De Nolf