2015-11-12 00:54:25 +01:00
|
|
|
import unittest
|
|
|
|
|
|
2021-01-31 16:49:12 -05:00
|
|
|
from circular_buffer import (
|
|
|
|
|
CircularBuffer,
|
2021-10-25 09:36:44 -07:00
|
|
|
BufferEmptyException,
|
|
|
|
|
BufferFullException,
|
2021-01-31 16:49:12 -05:00
|
|
|
)
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2020-10-15 12:46:24 -04:00
|
|
|
# Tests adapted from `problem-specifications//canonical-data.json`
|
2017-10-27 12:28:45 -02:00
|
|
|
|
2019-10-02 17:26:34 -04:00
|
|
|
|
2015-11-12 00:54:25 +01:00
|
|
|
class CircularBufferTest(unittest.TestCase):
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_reading_empty_buffer_should_fail(self):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf = CircularBuffer(1)
|
2021-10-25 09:36:44 -07:00
|
|
|
with self.assertRaises(BufferError) as err:
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.read()
|
2021-10-25 09:36:44 -07:00
|
|
|
self.assertEqual(
|
|
|
|
|
type(err.exception), (BufferEmptyException, BufferFullException)
|
|
|
|
|
)
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
err.exception.args[0],
|
|
|
|
|
("Circular buffer is empty", "Circular buffer is full"),
|
|
|
|
|
)
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_can_read_an_item_just_written(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
2017-10-27 12:28:45 -02:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_each_item_may_only_be_read_once(self):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf = CircularBuffer(1)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
2021-10-25 09:36:44 -07:00
|
|
|
with self.assertRaises(BufferError) as err:
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.read()
|
2021-10-25 09:36:44 -07:00
|
|
|
self.assertEqual(
|
|
|
|
|
type(err.exception), (BufferEmptyException, BufferFullException)
|
|
|
|
|
)
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
err.exception.args[0],
|
|
|
|
|
("Circular buffer is empty", "Circular buffer is full"),
|
|
|
|
|
)
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_items_are_read_in_the_order_they_are_written(self):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf = CircularBuffer(2)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
buf.write("2")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
|
|
|
|
self.assertEqual(buf.read(), "2")
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-10-02 17:26:34 -04:00
|
|
|
def test_full_buffer_can_t_be_written_to(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
2021-10-25 09:36:44 -07:00
|
|
|
with self.assertRaises(BufferError) as err:
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("2")
|
2021-10-25 09:36:44 -07:00
|
|
|
self.assertEqual(
|
|
|
|
|
type(err.exception), (BufferEmptyException, BufferFullException)
|
|
|
|
|
)
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
err.exception.args[0],
|
|
|
|
|
("Circular buffer is empty", "Circular buffer is full"),
|
|
|
|
|
)
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_a_read_frees_up_capacity_for_another_write(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
|
|
|
|
buf.write("2")
|
|
|
|
|
self.assertEqual(buf.read(), "2")
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-10-02 17:26:34 -04:00
|
|
|
def test_read_position_is_maintained_even_across_multiple_writes(self):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf = CircularBuffer(3)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
buf.write("2")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
|
|
|
|
buf.write("3")
|
|
|
|
|
self.assertEqual(buf.read(), "2")
|
|
|
|
|
self.assertEqual(buf.read(), "3")
|
|
|
|
|
|
|
|
|
|
def test_items_cleared_out_of_buffer_can_t_be_read(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.clear()
|
2021-10-25 09:36:44 -07:00
|
|
|
with self.assertRaises(BufferError) as err:
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.read()
|
2021-10-25 09:36:44 -07:00
|
|
|
self.assertEqual(
|
|
|
|
|
type(err.exception), (BufferEmptyException, BufferFullException)
|
|
|
|
|
)
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
err.exception.args[0],
|
|
|
|
|
("Circular buffer is empty", "Circular buffer is full"),
|
|
|
|
|
)
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_clear_frees_up_capacity_for_another_write(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.clear()
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("2")
|
|
|
|
|
self.assertEqual(buf.read(), "2")
|
2017-10-27 12:28:45 -02:00
|
|
|
|
2018-02-20 01:35:00 +08:00
|
|
|
def test_clear_does_nothing_on_empty_buffer(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
|
|
|
|
buf.clear()
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_overwrite_acts_like_write_on_non_full_buffer(self):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf = CircularBuffer(2)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
buf.overwrite("2")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
|
|
|
|
self.assertEqual(buf.read(), "2")
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_overwrite_replaces_the_oldest_item_on_full_buffer(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(2)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
buf.write("2")
|
|
|
|
|
buf.overwrite("3")
|
|
|
|
|
self.assertEqual(buf.read(), "2")
|
|
|
|
|
self.assertEqual(buf.read(), "3")
|
|
|
|
|
|
|
|
|
|
def test_overwrite_replaces_the_oldest_item_remaining_in_buffer_following_a_read(
|
2021-01-31 16:49:12 -05:00
|
|
|
self,
|
2019-10-02 17:26:34 -04:00
|
|
|
):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(3)
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
buf.write("2")
|
|
|
|
|
buf.write("3")
|
|
|
|
|
self.assertEqual(buf.read(), "1")
|
|
|
|
|
buf.write("4")
|
|
|
|
|
buf.overwrite("5")
|
|
|
|
|
self.assertEqual(buf.read(), "3")
|
|
|
|
|
self.assertEqual(buf.read(), "4")
|
|
|
|
|
self.assertEqual(buf.read(), "5")
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2019-08-04 17:23:46 +01:00
|
|
|
def test_initial_clear_does_not_affect_wrapping_around(self):
|
|
|
|
|
buf = CircularBuffer(2)
|
|
|
|
|
buf.clear()
|
2019-10-02 17:26:34 -04:00
|
|
|
buf.write("1")
|
|
|
|
|
buf.write("2")
|
|
|
|
|
buf.overwrite("3")
|
|
|
|
|
buf.overwrite("4")
|
|
|
|
|
self.assertEqual(buf.read(), "3")
|
|
|
|
|
self.assertEqual(buf.read(), "4")
|
2021-10-25 09:36:44 -07:00
|
|
|
with self.assertRaises(BufferError) as err:
|
2019-08-04 17:23:46 +01:00
|
|
|
buf.read()
|
2021-10-25 09:36:44 -07:00
|
|
|
self.assertEqual(
|
|
|
|
|
type(err.exception), (BufferEmptyException, BufferFullException)
|
|
|
|
|
)
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
err.exception.args[0],
|
|
|
|
|
("Circular buffer is empty", "Circular buffer is full"),
|
|
|
|
|
)
|