Coverage for tests/io/test_buffer.py: 100%

1"""EventAccumulator tests (evutils.io.buffer). 

2 

3The accumulator is the staging buffer behind the streaming pipeline generators. 

4The stream tests only push small data, so the capacity-pressure paths -- rotate 

5(reclaim consumed front), trigger-buffer grow, and the overflow guard -- are 

6covered here directly. 

7""" 

8import numpy as np 

9import pytest 

10 

11from evutils.types import EventArray, TriggerArray 

12from evutils.io.buffer import EventAccumulator 

13 

14 

15def ev(ts) -> EventArray: 

16 ts = list(ts) 

17 n = len(ts) 

18 return EventArray(t=ts, x=[0] * n, y=[0] * n, p=[0] * n) 

19 

20 

21def tr(ts) -> TriggerArray: 

22 ts = list(ts) 

23 n = len(ts) 

24 return TriggerArray(t=ts, p=[0] * n, id=list(range(n))) 

25 

26 

27def test_accumulator_slice_copy_advances(): 

28 acc = EventAccumulator(capacity=100) 

29 acc.append(ev([0, 1, 2, 3])) 

30 assert len(acc) == 4 

31 e, _ = acc.slice_copy(2, 0) 

32 assert e.t.tolist() == [0, 1] 

33 assert len(acc) == 2 

34 assert acc.t_window().tolist() == [2, 3] 

35 

36 

37def test_accumulator_slice_copy_is_independent(): 

38 acc = EventAccumulator(capacity=100) 

39 acc.append(ev([5, 6, 7])) 

40 e, _ = acc.slice_copy(3, 0) 

41 acc.reset() 

42 acc.append(ev([99, 99, 99])) 

43 assert e.t.tolist() == [5, 6, 7] # earlier copy untouched by later writes 

44 

45 

46def test_accumulator_rotate_events_and_triggers(): 

47 """Consuming the front then appending past the tail room triggers _rotate, 

48 which must preserve the unconsumed remainder of both events and triggers.""" 

49 acc = EventAccumulator(capacity=64) # trigger cap = 64 // 16 = 4 

50 acc.append(ev(range(8)), tr([0, 3, 6])) 

51 acc.slice_copy(5, 2) # consume 5 events, 2 triggers 

52 assert acc.t_window().tolist() == [5, 6, 7] 

53 

54 acc.append(ev(range(100, 160)), tr([200, 201])) # 64-8 < 60 -> rotate 

55 assert len(acc) == 63 # 3 remainder + 60 new 

56 assert acc.t_window()[:3].tolist() == [5, 6, 7] 

57 assert acc.t_window()[3:].tolist() == list(range(100, 160)) 

58 assert acc.t_window_tr().tolist() == [6, 200, 201] 

59 

60 

61def test_accumulator_rotate_after_full_consume(): 

62 """Rotate when everything is consumed: the remainder is zero, so _rotate 

63 resets the offsets without copying (the n==0 / n_tr==0 branches).""" 

64 acc = EventAccumulator(capacity=10) 

65 acc.append(ev(range(6)), tr([1, 2])) 

66 acc.slice_copy(6, 2) # consume all events and triggers 

67 assert len(acc) == 0 

68 acc.append(ev(range(100, 108))) # 10-6 < 8 -> rotate, nothing to move 

69 assert acc.t_window().tolist() == list(range(100, 108)) 

70 

71 

72def test_accumulator_overflow_raises(): 

73 """Appending more than the total capacity (even after a rotate) errors.""" 

74 acc = EventAccumulator(capacity=5) 

75 acc.append(ev(range(4))) 

76 with pytest.raises(ValueError, match="full"): 

77 acc.append(ev(range(4))) 

78 

79 

80def test_accumulator_trigger_buffer_grows(): 

81 """More triggers than the trigger buffer's capacity trigger a grow().""" 

82 acc = EventAccumulator(capacity=32) # trigger cap = 32 // 16 = 2 

83 acc.append(ev([0]), tr([1, 2, 3, 4, 5])) # 5 > 2 -> grow 

84 assert acc.t_window_tr().tolist() == [1, 2, 3, 4, 5] 

85 

86 

87def test_accumulator_prepare_rotates_for_headroom(): 

88 acc = EventAccumulator(capacity=10) 

89 acc.append(ev(range(8))) 

90 acc.slice_copy(5, 0) # start=5, 3 unconsumed 

91 b, _ = acc.prepare(5) # 10-8 < 5 -> rotate 

92 assert acc.t_window().tolist() == [5, 6, 7] 

93 assert b.c.capacity == 8 # min(capacity, size + step) = min(10, 3+5) 

94 

95 

96def test_accumulator_reset_clears(): 

97 acc = EventAccumulator(capacity=16) 

98 acc.append(ev([1, 2, 3]), tr([1])) 

99 acc.reset() 

100 assert len(acc) == 0 

101 assert acc.t_window().tolist() == [] 

102 assert acc.t_window_tr().tolist() == []