2015-11-12 00:54:25 +01:00
|
|
|
import unittest
|
|
|
|
|
|
2016-08-17 11:57:29 -07:00
|
|
|
from circular_buffer import (
|
|
|
|
|
CircularBuffer,
|
|
|
|
|
BufferFullException,
|
|
|
|
|
BufferEmptyException
|
|
|
|
|
)
|
2015-11-12 00:54:25 +01:00
|
|
|
|
|
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
# Tests adapted from `problem-specifications//canonical-data.json` @ v1.0.1
|
|
|
|
|
|
2015-11-12 00:54:25 +01:00
|
|
|
class CircularBufferTest(unittest.TestCase):
|
|
|
|
|
def test_read_empty_buffer(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
2017-12-12 18:11:43 +00:00
|
|
|
with self.assertRaisesWithMessage(BufferEmptyException):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.read()
|
|
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
def test_read_just_written_item(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
|
|
|
|
buf.write('1')
|
|
|
|
|
self.assertEqual(buf.read(), '1')
|
|
|
|
|
|
2015-11-12 00:54:25 +01:00
|
|
|
def test_write_and_read_back_one_item(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
|
|
|
|
buf.write('1')
|
2017-10-27 12:28:45 -02:00
|
|
|
self.assertEqual(buf.read(), '1')
|
2017-12-12 18:11:43 +00:00
|
|
|
with self.assertRaisesWithMessage(BufferEmptyException):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.read()
|
|
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
def test_write_and_read_back_multiple_items_ordered(self):
|
2015-11-12 00:54:25 +01:00
|
|
|
buf = CircularBuffer(2)
|
|
|
|
|
buf.write('1')
|
|
|
|
|
buf.write('2')
|
2017-03-23 13:37:20 +01:00
|
|
|
self.assertEqual(buf.read(), '1')
|
|
|
|
|
self.assertEqual(buf.read(), '2')
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
def test_full_buffer_cant_written(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('1')
|
2017-12-12 18:11:43 +00:00
|
|
|
with self.assertRaisesWithMessage(BufferFullException):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.write('2')
|
2015-11-12 00:54:25 +01:00
|
|
|
|
|
|
|
|
def test_alternate_write_and_read(self):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf = CircularBuffer(1)
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('1')
|
2017-03-23 13:37:20 +01:00
|
|
|
self.assertEqual(buf.read(), '1')
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('2')
|
2017-03-23 13:37:20 +01:00
|
|
|
self.assertEqual(buf.read(), '2')
|
2015-11-12 00:54:25 +01:00
|
|
|
|
|
|
|
|
def test_read_back_oldest_item(self):
|
|
|
|
|
buf = CircularBuffer(3)
|
|
|
|
|
buf.write('1')
|
|
|
|
|
buf.write('2')
|
2017-10-27 12:28:45 -02:00
|
|
|
self.assertEqual(buf.read(), '1')
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('3')
|
2017-10-27 12:28:45 -02:00
|
|
|
self.assertEqual(buf.read(), '2')
|
2017-03-23 13:37:20 +01:00
|
|
|
self.assertEqual(buf.read(), '3')
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
def test_clearing_buffer(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('1')
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.clear()
|
2017-12-12 18:11:43 +00:00
|
|
|
with self.assertRaisesWithMessage(BufferEmptyException):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.read()
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
def test_clear_free_buffer_for_write(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('1')
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.clear()
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('2')
|
2017-03-23 13:37:20 +01:00
|
|
|
self.assertEqual(buf.read(), '2')
|
2017-10-27 12:28:45 -02:00
|
|
|
|
|
|
|
|
def test_clear_does_nothin_empty_buffer(self):
|
|
|
|
|
buf = CircularBuffer(1)
|
|
|
|
|
buf.clear()
|
|
|
|
|
buf.write('1')
|
|
|
|
|
self.assertEqual(buf.read(), '1')
|
2015-11-12 00:54:25 +01:00
|
|
|
|
|
|
|
|
def test_overwrite_non_full_buffer(self):
|
|
|
|
|
buf = CircularBuffer(2)
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.write('1')
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.overwrite('2')
|
2017-03-23 13:37:20 +01:00
|
|
|
self.assertEqual(buf.read(), '1')
|
|
|
|
|
self.assertEqual(buf.read(), '2')
|
2015-11-12 00:54:25 +01:00
|
|
|
|
2017-10-27 12:28:45 -02:00
|
|
|
def test_overwrite_replaces_oldest_item(self):
|
|
|
|
|
buf = CircularBuffer(2)
|
|
|
|
|
buf.write('1')
|
|
|
|
|
buf.write('2')
|
|
|
|
|
buf.overwrite('3')
|
|
|
|
|
self.assertEqual(buf.read(), '2')
|
|
|
|
|
self.assertEqual(buf.read(), '3')
|
|
|
|
|
|
|
|
|
|
def test_write_full_buffer(self):
|
|
|
|
|
buf = CircularBuffer(2)
|
|
|
|
|
buf.write('1')
|
|
|
|
|
buf.write('2')
|
2017-12-12 18:11:43 +00:00
|
|
|
with self.assertRaisesWithMessage(BufferFullException):
|
2017-10-27 12:28:45 -02:00
|
|
|
buf.write('A')
|
|
|
|
|
|
|
|
|
|
def test_over_write_replaces_oldest_remaning_item(self):
|
|
|
|
|
buf = CircularBuffer(3)
|
|
|
|
|
buf.write('1')
|
|
|
|
|
buf.write('2')
|
|
|
|
|
buf.write('3')
|
|
|
|
|
self.assertEqual(buf.read(), '1')
|
2015-11-12 00:54:25 +01:00
|
|
|
buf.write('4')
|
2017-10-27 12:28:45 -02:00
|
|
|
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
|
|
|
|
2017-12-12 18:11:43 +00:00
|
|
|
# Utility functions
|
|
|
|
|
def setUp(self):
|
|
|
|
|
try:
|
2018-01-18 10:47:11 -06:00
|
|
|
self.assertRaisesRegex
|
2017-12-12 18:11:43 +00:00
|
|
|
except AttributeError:
|
2018-01-18 10:47:11 -06:00
|
|
|
self.assertRaisesRegex = self.assertRaisesRegexp
|
2017-12-12 18:11:43 +00:00
|
|
|
|
|
|
|
|
def assertRaisesWithMessage(self, exception):
|
|
|
|
|
return self.assertRaisesRegex(exception, r".+")
|
|
|
|
|
|
2015-11-12 00:54:25 +01:00
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
|
|
unittest.main()
|