Coverage for tests/io/test_phase1_regressions.py: 100%
95 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Regression tests for the Phase-1 correctness fixes.
3Each test pins one previously-broken behaviour:
4- tiny chunk_size used to underflow the C parsers' capacity headroom
5 (size_t wrap => heap corruption hazard);
6- delta_t fast-path overshoot (``_dt_carry``) used to be invisible to
7 mixed-mode reads and ``read_all()`` (silent event loss);
8- ``max_events`` was not actually enforced within one C delta_t call;
9- ``EventWriter.write()`` silently dropped its ``triggers`` argument;
10- EVT2.1 with a non-legacy endianness header decoded silently into garbage.
11"""
12import numpy as np
13import pytest
15from typing import Any
18def _uniform_events(n: int = 5000, dt: int = 13) -> Any:
19 ev = np.zeros(n, dtype=np.dtype([('t', np.int64), ('x', np.uint16),
20 ('y', np.uint16), ('p', np.uint8)]))
21 ev['t'] = np.arange(n, dtype=np.int64) * dt
22 ev['x'] = np.arange(n) % 1280
23 ev['y'] = np.arange(n) % 720
24 ev['p'] = np.arange(n) % 2
25 return ev
28def _write(tmp_path: Any, fmt: str, ev: Any) -> Any:
29 from evutils.io import EventWriter
30 p = tmp_path / f"reg_{fmt}.raw"
31 with EventWriter(p, format=fmt) as w:
32 w.write(ev)
33 return p
36####################################
37# C1: tiny chunk_size
38####################################
40def test_tiny_chunk_size_rejected(tmp_path: Any) -> None:
41 from evutils.io import EventReader
42 p = _write(tmp_path, "evt3", _uniform_events(100))
43 with pytest.raises(ValueError, match="chunk_size"):
44 EventReader(p, n_events=10, chunk_size=16).read()
47def test_minimal_chunk_size_decodes(tmp_path: Any) -> None:
48 """chunk_size=128 (the minimum) must decode the whole stream, not hang or
49 corrupt: the parser's capacity headroom (64) stays below the buffer size."""
50 from evutils.io import EventReader
51 ev = _uniform_events(1000)
52 p = _write(tmp_path, "evt3", ev)
53 total = 0
54 with EventReader(p, n_events=200, chunk_size=128) as r:
55 for c in r:
56 total += len(c)
57 assert total == len(ev)
60####################################
61# C3: carry visibility on mixed paths
62####################################
64@pytest.mark.parametrize("fmt", ["evt2", "evt3"])
65def test_mixed_delta_t_then_read_all_loses_nothing(tmp_path: Any, fmt: str) -> None:
66 from evutils.io import EventReader
67 ev = _uniform_events(5000)
68 p = _write(tmp_path, fmt, ev)
69 got = []
70 with EventReader(p, delta_t=10_000) as r:
71 # a few fast-path windows...
72 for _ in range(3):
73 got.append(np.asarray(r.read()).copy())
74 # ...then drain the rest at once: the fast-path overshoot carry must be
75 # folded in, not dropped.
76 got.append(np.asarray(r.read_all()).copy())
77 out = np.concatenate([g for g in got if len(g)])
78 assert len(out) == len(ev)
79 assert np.array_equal(out['t'], ev['t'])
82@pytest.mark.parametrize("fmt", ["evt2", "evt3"])
83def test_mixed_delta_t_then_n_events_override(tmp_path: Any, fmt: str) -> None:
84 from evutils.io import EventReader
85 ev = _uniform_events(5000)
86 p = _write(tmp_path, fmt, ev)
87 got = []
88 with EventReader(p, delta_t=10_000) as r:
89 got.append(np.asarray(r.read()).copy()) # fast path window
90 got.append(np.asarray(r.read(n_events=700)).copy()) # override -> acc path
91 while True: # drain
92 c = np.asarray(r.read())
93 if len(c) == 0:
94 break
95 got.append(c.copy())
96 out = np.concatenate([g for g in got if len(g)])
97 assert len(out) == len(ev)
98 assert np.array_equal(out['t'], ev['t'])
101####################################
102# C4: max_events honoured in delta_t mode
103####################################
105def test_delta_t_windows_respect_max_events(tmp_path: Any) -> None:
106 from evutils.io import EventReader
107 ev = _uniform_events(5000)
108 p = _write(tmp_path, "evt3", ev)
109 total = 0
110 # One delta_t window would span the whole file; the cap must split it.
111 with EventReader(p, delta_t=10_000_000, max_events=1000) as r:
112 for c in r:
113 assert len(c) <= 1000
114 total += len(c)
115 assert total == len(ev)
118####################################
119# C5: EventWriter.write forwards triggers
120####################################
122def test_writer_warns_on_unsupported_triggers(tmp_path: Any) -> None:
123 """No encoder implements trigger *encoding* yet, so the honest contract is:
124 EventWriter forwards triggers to the encoder and warns LOUDLY (once) when
125 the encoder cannot store them -- never a silent drop. When an encoder gains
126 trigger support (SUPPORTS_WRITE_TRIGGERS = True) this becomes a real
127 round-trip test."""
128 from evutils.io import EventReader, EventWriter
129 from evutils.types import TriggerArray
130 ev = _uniform_events(2000)
131 tr = TriggerArray(t=[100, 5000, 20_000], p=[1, 0, 1], id=[0, 1, 0])
132 p = tmp_path / "trig.raw"
133 with EventWriter(p, format="evt3") as w:
134 with pytest.warns(UserWarning, match="triggers"):
135 w.write(ev, triggers=tr)
136 w.write(ev, triggers=tr) # warning fires only once
137 with EventReader(p) as r:
138 out = r.read_all()
139 assert len(out) == 2 * len(ev) # events themselves are unaffected
142####################################
143# C9: EVT2.1 endianness guard
144####################################
146def test_evt21_non_legacy_endianness_rejected() -> None:
147 from evutils.io import EventReader
148 hdr = b"% evt 2.1\n% endianness little\n% geometry 1280x720\n% end\n"
149 with pytest.raises(NotImplementedError, match="endianness"):
150 EventReader(hdr).read()
153def test_evt21_legacy_endianness_accepted(tmp_path: Any) -> None:
154 from evutils.io import EventReader
155 ev = _uniform_events(500)
156 p = _write(tmp_path, "evt21", ev) # writer emits "% endianness legacy"
157 with EventReader(p) as r:
158 out = r.read_all()
159 assert len(out) == len(ev)