diff options
| -rw-r--r-- | src/argaze.test/DataStructures.py | 143 |
1 files changed, 143 insertions, 0 deletions
diff --git a/src/argaze.test/DataStructures.py b/src/argaze.test/DataStructures.py index cab3a31..5f0a4ac 100644 --- a/src/argaze.test/DataStructures.py +++ b/src/argaze.test/DataStructures.py @@ -4,6 +4,29 @@ import unittest from argaze import DataStructures +def random_data_buffer(size, data_keys): + """ Generate a random TimeStampedBuffer for testing purpose. + Timestamps are current time. + Values are tuples containing an expected value and a random value. + """ + + import random + import time + + ts_buffer = DataStructures.TimeStampedBuffer() + + for i in range(0, size): + + # Edit data + random_data = {} + for key in data_keys: + random_data[key] = (i, random.random()) + + # Store data + ts_buffer[time.time()] = random_data + + return ts_buffer + class TestTimeStampedBufferClass(unittest.TestCase): """Test TimeStampedBuffer class.""" @@ -49,6 +72,126 @@ class TestTimeStampedBufferClass(unittest.TestCase): self.assertEqual(len(ts_buffer.append(ts_buffer_next)), 4) self.assertEqual(list(ts_buffer.append(ts_buffer_next).keys()), [0, 1, 2, 3]) + def test_first(self): + """Test TimeStampedBuffer first property.""" + + self.assertEqual(DataStructures.TimeStampedBuffer({0: "A", 1: "B"}).first, (0, "A")) + + # Check that accessing to first item of an empty TimeStampedBuffer fails + with self.assertRaises(IndexError): + + DataStructures.TimeStampedBuffer().first + + def test_pop_first(self): + """Test TimeStampedBuffer pop_first method.""" + + ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B"}) + + self.assertEqual(ts_buffer.pop_first(), (0, "A")) + self.assertEqual(len(ts_buffer), 1) + self.assertEqual(ts_buffer.first, (1, "B")) + + def test_pop_first_until(self): + """Test TimeStampedBuffer pop_first_until method.""" + + ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + + # Check pop until an existing timestamp + first_until_2 = ts_buffer.pop_first_until(2) + + self.assertEqual(first_until_2, (1, "B")) + self.assertEqual(len(ts_buffer), 2) + self.assertEqual(ts_buffer.first, (2, "C")) + + # Check pop until an none existing timestamp + ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + + first_until_1dot5 = ts_buffer.pop_first_until(1.5) + + self.assertEqual(first_until_1dot5, (1, "B")) + self.assertEqual(len(ts_buffer), 2) + self.assertEqual(ts_buffer.first, (2, "C")) + + def test_last(self): + """Test TimeStampedBuffer last property.""" + + self.assertEqual(DataStructures.TimeStampedBuffer({0: "A", 1: "B"}).last, (1, "B")) + + # Check that accessing to last item of an empty TimeStampedBuffer fails + with self.assertRaises(IndexError): + + DataStructures.TimeStampedBuffer().last + + def test_pop_last(self): + """Test TimeStampedBuffer pop_last method.""" + + ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B"}) + + self.assertEqual(ts_buffer.pop_last(), (1, "B")) + self.assertEqual(len(ts_buffer), 1) + self.assertEqual(ts_buffer.last, (0, "A")) + + def test_get_last_before(self): + """Test TimeStampedBuffer get_last_before method.""" + + ts_buffer = DataStructures.TimeStampedBuffer({0: "A", 1: "B", 2: "C", 3: "D"}) + + get_last_before_2 = ts_buffer.get_last_before(2) + + self.assertEqual(get_last_before_2, (1, "B")) + self.assertEqual(len(ts_buffer), 4) + + get_last_before_1dot5 = ts_buffer.get_last_before(1.5) + + self.assertEqual(get_last_before_1dot5, (1, "B")) + + get_last_before_4 = ts_buffer.get_last_before(4) + + self.assertEqual(get_last_before_4, (3, "D")) + + # Check that accessing to early timestamp fails + with self.assertRaises(KeyError): + + ts_buffer.get_last_before(-1) + + def test_as_dataframe(self): + """Test TimeStampedBuffer as_dataframe method.""" + + ts_buffer = random_data_buffer(10, ["data_A", "data_B", "data_C"]) + + # Check dataframe conversion + ts_buffer_dataframe = ts_buffer.as_dataframe() + + self.assertEqual(ts_buffer_dataframe.index.name, "timestamp") + self.assertEqual(ts_buffer_dataframe.index.size, 10) + + self.assertEqual(ts_buffer_dataframe.columns.size, 3) + self.assertEqual(ts_buffer_dataframe.columns[0], "data_A") + self.assertEqual(ts_buffer_dataframe.columns[1], "data_B") + self.assertEqual(ts_buffer_dataframe.columns[2], "data_C") + + # Check data exclusion option + ts_buffer_dataframe = ts_buffer.as_dataframe(exclude=["data_B"]) + + self.assertEqual(ts_buffer_dataframe.index.name, "timestamp") + self.assertEqual(ts_buffer_dataframe.index.size, 10) + + self.assertEqual(ts_buffer_dataframe.columns.size, 2) + self.assertEqual(ts_buffer_dataframe.columns[0], "data_A") + self.assertEqual(ts_buffer_dataframe.columns[1], "data_C") + + # Check dataframe split option + ts_buffer_dataframe = ts_buffer.as_dataframe(split={"data_B": ["data_B0", "data_B1"]}) + + self.assertEqual(ts_buffer_dataframe.index.name, "timestamp") + self.assertEqual(ts_buffer_dataframe.index.size, 10) + + self.assertEqual(ts_buffer_dataframe.columns.size, 4) + self.assertEqual(ts_buffer_dataframe.columns[0], "data_A") + self.assertEqual(ts_buffer_dataframe.columns[1], "data_B0") + self.assertEqual(ts_buffer_dataframe.columns[2], "data_B1") + self.assertEqual(ts_buffer_dataframe.columns[3], "data_C") + if __name__ == '__main__': unittest.main()
\ No newline at end of file |