diff --git a/tests/unit/byteb4rb1e_utils/collections/__init__.py b/tests/unit/byteb4rb1e_utils/collections/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/byteb4rb1e_utils/collections/test_circular_buffer.py b/tests/unit/byteb4rb1e_utils/collections/test_circular_buffer.py new file mode 100644 index 0000000..aeb0934 --- /dev/null +++ b/tests/unit/byteb4rb1e_utils/collections/test_circular_buffer.py @@ -0,0 +1,82 @@ +import unittest + +from byteb4rb1e_utils.collections import CircularBuffer + +class test_init(unittest.TestCase): + """CircularBuffer.__init__()""" + + def test_default(self): + """Test buffer initializes correctly.""" + buffer = CircularBuffer(10) + self.assertEqual(len(buffer.buf), 10) + self.assertEqual(buffer.size, 10) + self.assertEqual(buffer.start, 0) + self.assertEqual(buffer.end, 0) + self.assertFalse(buffer.filled) + + +class test_append(unittest.TestCase): + """CircularBuffer.append()""" + + def test_without_overflow(self): + """Test appending bytes without overwriting.""" + buffer = CircularBuffer(5) + buffer.append(b"abc") + self.assertEqual(buffer.buf[:3], bytearray(b"abc")) + self.assertEqual(buffer.start, 0) + self.assertEqual(buffer.end, 3) + self.assertFalse(buffer.filled) + + def test_with_overflow(self): + """appending bytes with overwriting""" + buffer = CircularBuffer(5) + buffer.append(b"abcde") # Fill buffer completely + buffer.append(b"fg") # Overwrite some elements + + self.assertEqual(buffer.buf, bytearray(b"fgcde")) # Last five elements should remain + self.assertEqual(buffer.start, 3) # Should have moved forward due to overwrite + self.assertEqual(buffer.end, 2) # Wrapped around + self.assertTrue(buffer.filled) # Buffer should indicate overwrite occurred + + def test_wraparound_behavior(self): + """correct handling of buffer wraparound""" + buffer = CircularBuffer(4) + buffer.append(b"abcd") # Fill buffer completely + buffer.append(b"e") # Overwrite first element + + self.assertEqual(buffer.buf, bytearray(b"ebcd")) # Should have wrapped around + self.assertEqual(buffer.start, 2) + self.assertEqual(buffer.end, 1) + self.assertTrue(buffer.filled) + + def test_consecutive_overwrites(self): + """repeated buffer overwrites across multiple cycles + + verifies that the circular buffer correctly handles multiple overwrite cycles. + + - Initially, the buffer is filled completely (`abcde`). + - New data (`fghij`) is appended, causing the oldest elements to be replaced. + - The buffer does not physically shift; only the `start` index moves forward as data is overwritten. + - When wrapping around, the `start` advances while `end` cycles back to zero. + + Expected: + - Buffer contents should be `bytearray(b"fghij")`, containing only + the most recent data + - `start` moves forward due to full overwriting. + - `end` correctly wraps around. + - `filled` flag indicates that overwrites have occurred. + + **Why `start = 1` instead of `0`?** + Well it took some brain juice for me to fully comprehend it, but in a + circular buffer, `start` always marks the oldest remaining element. + When a full overwrite occurs, `start` moves forward to indicate the next + oldest position, ensuring sequential ordering is preserved. + """ + buffer = CircularBuffer(5) + buffer.append(b"abcde") # Fill buffer + buffer.append(b"fghij") # Overwrite completely + + self.assertEqual(buffer.buf, bytearray(b"fghij")) # Only the latest data remains + self.assertEqual(buffer.start, 1) + self.assertEqual(buffer.end, 0) + self.assertTrue(buffer.filled)