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"]