v1.7.1: accept EOFError as clean end-of-stream (newer libsigrokdecode)

Newer libsigrokdecode signals end-of-capture from wait() by raising
EOFError("samples exhausted"); Linux 0.5.3 instead returns None / raises
SystemError, so this only reproduced on Windows.  The framework explicitly
accepts EOFError out of decode() as normal termination, but decode() caught
it in the generic except Exception handler -- printing a traceback, logging
DECODE ERROR, and resyncing up to 50 times, which flooded the log at the end
of every capture.

- decode() and the five "except SystemError: raise" guards now use
  "except (SystemError, EOFError): raise", letting the end-of-stream signal
  propagate for a clean exit.  The critical guard is
  _measure_clock_period(), whose try wraps a wait() loop.
- 2 new unit tests assert decode() re-raises EOFError and SystemError.
- 67/67 unit tests; test_8/test_7/xiaomi_phone unchanged (Linux path).
  VERSION -> 1.7.1.
This commit is contained in:
Трошин Антон Валерьевич committed 2026-09-11 18:12:03 +03:00
1 parent 690cc132ca
commit 7db29e3ce2
3 files changed
+70 -12

No files matched your search

+32
View File
@@ -530,5 +530,37 @@ class TestAtEof(unittest.TestCase):
self.assertTrue(self.inst._at_eof())
class TestEndOfStream(unittest.TestCase):
"""decode() must let end-of-stream exceptions propagate.
Newer libsigrokdecode signals end-of-capture from wait() with
EOFError('samples exhausted') (older versions use SystemError).
libsigrokdecode accepts either as normal decode() termination, so
swallowing them turns a clean end into a flood of DECODE ERROR retries.
"""
def setUp(self):
_, pd = _load_iso7816()
self.pd = pd
def _decode_raising(self, exc_cls):
inst = object.__new__(self.pd.Decoder)
inst.write_pcap_header = lambda: None
inst.finish = lambda: None
def step():
raise exc_cls('samples exhausted')
inst.decode_step = step
return inst
def test_eoferror_propagates(self):
with self.assertRaises(EOFError):
self._decode_raising(EOFError).decode()
def test_systemerror_propagates(self):
with self.assertRaises(SystemError):
self._decode_raising(SystemError).decode()
if __name__ == '__main__':
unittest.main()