import pytest import numpy as np from IPython import embed from pyrelacs.dataio.circbuffer import CircBuffer def test_init(): buff = CircBuffer(1000, 2) assert buff.size == 1000 assert buff.channel_count == 2 def test_hasvalue(): buff = CircBuffer(1000, 2) assert buff.has_value(0, 0) == False assert buff.has_value(-1, 0) == False buff.append(10, 0) assert buff.write_index(0) == 1 assert buff.write_index(1) == 0 assert buff.has_value(0, 0) == True assert buff.has_value(0, 1) == False buff.append(10, 1) assert buff.write_index(1) == 1 assert buff.has_value(0, 1) == True for i in range(1100): buff.append(i, 0) buff.append(i, 1) assert buff.write_index(0) == buff.write_index(1) assert buff.has_value(0, 0) == True assert buff.has_value(0, 1) == True assert buff.has_value(buff.write_index(0), 0) == True assert buff.has_value(buff.write_index(1), 1) == True def test_validrange(): buff = CircBuffer(1000, 2) # without any values the range is (0, 0) assert buff.valid_range() == (0, 0) buff.append(0, 0) assert buff.valid_range() == (0, 1) for i in range(100): buff.append(i, 0) assert buff.valid_range() == (0, 101) for i in range(1000): buff.append(i, 0) assert buff.valid_range() == (0, 1000) def test_get(): buff = CircBuffer(1000, 2) # with no items written to the buffer with pytest.raises(IndexError): item = buff.get(index=-1) buff.append(10, 0) item = buff.get(index=-1) assert item == 10 # Check if index is not written jet with pytest.raises(IndexError): item = buff.get(index=10) for i in range(1000): buff.append(i, 0) item = buff.get(index=-1) # the first item should be 999.0 because of we append a value in the earlier test assert item == 999.0 with pytest.raises(IndexError): item = buff.get(10001) def test_read(): pass def test_write(): buff = CircBuffer(1000, 2) samplecount = 1000 if __name__ == "__main__": test_get()