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_grows_past_capacity(): 

73 """Appending more than the current capacity grows the buffer (rather than 

74 erroring) so oversized windows and drain-to-EOF reads always fit.""" 

75 acc = EventAccumulator(capacity=5) 

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

77 acc.append(ev(range(4))) # 8 > capacity 5 -> grow, no error 

78 assert len(acc) == 8 

79 assert acc.t_window().tolist() == [0, 1, 2, 3, 0, 1, 2, 3] 

80 

81 

82def test_accumulator_trigger_buffer_grows(): 

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

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

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

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

87 

88 

89def test_accumulator_prepare_rotates_for_headroom(): 

90 acc = EventAccumulator(capacity=10) 

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

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

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

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

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

96 

97 

98def test_accumulator_reset_clears(): 

99 acc = EventAccumulator(capacity=16) 

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

101 acc.reset() 

102 assert len(acc) == 0 

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

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