f53e2ef08b
* Implement checks for messages being raised with exceptions (Fixes #1080) * Add self.assertRaisesWithMessage method to relevant exercise tests - Uses self.assertRaisesRegex - Checks only for the presence of a message, not content * Add meaningful messages to failing examples * octal: Switch to using a context manager for exception tests * Add note regarding error messages to the insert * simple-linked-list: Move hints.md to correct location * simple-cipher: Remove extra whitespace from lines * collatz-conjecture: Update hints.md * Regenerate README to include exceptions section
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
class BufferFullException(Exception):
|
|
pass
|
|
|
|
|
|
class BufferEmptyException(Exception):
|
|
pass
|
|
|
|
|
|
class CircularBuffer(object):
|
|
|
|
def __init__(self, capacity):
|
|
self.buffer = bytearray(capacity)
|
|
self.read_point = 0
|
|
self.write_point = 0
|
|
|
|
# (protected) helper method to support python 2/3
|
|
def _update_buffer(self, data):
|
|
try:
|
|
self.buffer[self.write_point] = data
|
|
except TypeError:
|
|
self.buffer[self.write_point] = ord(data)
|
|
|
|
def clear(self):
|
|
self.buffer = bytearray(len(self.buffer))
|
|
|
|
def write(self, data):
|
|
if all(self.buffer):
|
|
raise BufferFullException("Circular buffer is full")
|
|
self._update_buffer(data)
|
|
self.write_point = (self.write_point + 1) % len(self.buffer)
|
|
|
|
def overwrite(self, data):
|
|
self._update_buffer(data)
|
|
if all(self.buffer) and self.write_point == self.read_point:
|
|
self.read_point = (self.read_point + 1) % len(self.buffer)
|
|
self.write_point = (self.write_point + 1) % len(self.buffer)
|
|
|
|
def read(self):
|
|
if not any(self.buffer):
|
|
raise BufferEmptyException("Circular buffer is empty")
|
|
data = chr(self.buffer[self.read_point])
|
|
self.buffer[self.read_point] = 0
|
|
self.read_point = (self.read_point + 1) % len(self.buffer)
|
|
return data
|