25 lines
617 B
Python
25 lines
617 B
Python
from plugins.host_agent.agent import RingBuffer
|
|
|
|
|
|
def test_ring_buffer_preserves_order():
|
|
rb = RingBuffer(maxlen=3)
|
|
rb.push(1); rb.push(2); rb.push(3)
|
|
assert list(rb.drain()) == [1, 2, 3]
|
|
assert len(rb) == 0
|
|
|
|
|
|
def test_ring_buffer_drops_oldest_when_full():
|
|
rb = RingBuffer(maxlen=3)
|
|
for v in (1, 2, 3, 4, 5):
|
|
rb.push(v)
|
|
assert list(rb.drain()) == [3, 4, 5]
|
|
|
|
|
|
def test_ring_buffer_drain_clears_and_is_atomic():
|
|
rb = RingBuffer(maxlen=5)
|
|
rb.push("a"); rb.push("b")
|
|
out = list(rb.drain())
|
|
rb.push("c")
|
|
assert out == ["a", "b"]
|
|
assert list(rb.drain()) == ["c"]
|