Commit 5cd74245 authored by Federico Vaga's avatar Federico Vaga

py: fix flush and test it

Signed-off-by: Federico Vaga's avatarFederico Vaga <federico.vaga@cern.ch>
parent 2948417d
......@@ -462,8 +462,7 @@ class FmcFineDelay(object):
def flush(self):
while True:
try:
ret = self.read(10, os.O_NONBLOCK)
if ret == 0:
if len(self.read(1024, os.O_NONBLOCK)) == 0:
break
except OSError:
break
......
......@@ -48,6 +48,15 @@ class TestFmcfdLoop(object):
chan 4 o-------`
"""
def test_output_flush(self, fmcfd, fmcfd_chan, fmcfd_tdc):
poll = select.poll()
poll.register(fmcfd_tdc.fileno, select.POLLIN)
fmcfd_chan.pulse_generate(fmcfd.time + FmcFineDelayTime(2, 0, 0),
200000, 400000, 16)
assert len(poll.poll(4000)) > 0
fmcfd_tdc.flush()
assert len(poll.poll(4000)) ==0
@pytest.mark.parametrize("count", [1, 2, 3, 5, 7, 10,
100, 1000, 10000, 65535])
def test_output_counter(self, fmcfd, fmcfd_chan, fmcfd_tdc, count):
......
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