Coverage for tests/io/test_fuzz_parsers.py: 94%

1import os 

2import random 

3from pathlib import Path 

4import pytest 

5from evutils.io import EventReader, EventWriter 

6from evutils.types import Event_dtype 

7import numpy as np 

8 

9FORMATS = [("evt3", "raw"), ("evt4", "raw"), ("evt2", "raw"), ("evt21", "raw"), ("dat", "dat"), ("csv", "csv")] 

10 

11def _write_valid_file(path, fmt): 

12 ev = np.zeros(10, dtype=Event_dtype) 

13 ev["t"] = np.arange(10) * 1000 

14 if fmt in ("evt3", "evt4", "evt2", "evt21"): 

15 with EventWriter(path, format=fmt) as w: 

16 w.write(ev) 

17 else: 

18 with EventWriter(path, width=1280, height=720) as w: 

19 w.write(ev) 

20 

21@pytest.mark.parametrize("fmt,ext", FORMATS) 

22@pytest.mark.filterwarnings("ignore:Malformed packets ignored near word:UserWarning") 

23def test_fuzz_parser_garbage(tmp_path, fmt, ext): 

24 """Feed random bytes after a valid header to ensure parsers don't segfault.""" 

25 p = tmp_path / f"fuzz_{fmt}.{ext}" 

26 _write_valid_file(p, fmt) 

27 

28 content = bytearray(p.read_bytes()) 

29 

30 # Append 1MB of random bytes 

31 random.seed(42) 

32 garbage = bytearray(random.getrandbits(8) for _ in range(100_000)) 

33 content.extend(garbage) 

34 

35 p.write_bytes(content) 

36 

37 # Read it back. We expect it to either return garbage events, raise an error, or stop. 

38 # It must NOT segfault or hang indefinitely. 

39 with EventReader(p, mode="all") as r: 

40 try: 

41 r.read() 

42 except Exception: 

43 pass # Parsing errors are acceptable, segfaults are not