Test file reader.
(self)
| 15 | |
| 16 | class TestFileBuffer(unittest.IsolatedAsyncioTestCase): |
| 17 | async def test_file_reader(self): # noqa: C901 |
| 18 | """Test file reader.""" |
| 19 | reader = get_buffer_reader(self.config.buffer.explorer_input.tasksets[0]) |
| 20 | |
| 21 | tasks = [] |
| 22 | while True: |
| 23 | try: |
| 24 | tasks.extend(await reader.read()) |
| 25 | except StopAsyncIteration: |
| 26 | break |
| 27 | self.assertEqual(len(tasks), 16) |
| 28 | |
| 29 | # test epoch and offset |
| 30 | self.config.buffer.explorer_input.tasksets[0].total_epochs = 2 |
| 31 | self.config.buffer.explorer_input.tasksets[0].index = 4 |
| 32 | reader = get_buffer_reader( |
| 33 | self.config.buffer.explorer_input.tasksets[0], |
| 34 | ) |
| 35 | tasks = [] |
| 36 | while True: |
| 37 | try: |
| 38 | tasks.extend(await reader.read()) |
| 39 | except StopAsyncIteration: |
| 40 | break |
| 41 | self.assertEqual(len(tasks), 16 * 2 - 4) |
| 42 | |
| 43 | # test total steps and offset |
| 44 | self.config.buffer.explorer_input.tasksets[0].total_steps = 5 |
| 45 | self.config.buffer.explorer_input.tasksets[0].index = 8 |
| 46 | reader = get_buffer_reader(self.config.buffer.explorer_input.tasksets[0]) |
| 47 | tasks = [] |
| 48 | while True: |
| 49 | try: |
| 50 | tasks.extend(await reader.read()) |
| 51 | except StopAsyncIteration: |
| 52 | break |
| 53 | self.assertEqual(len(tasks), 20 - 8) |
| 54 | |
| 55 | # test offset > dataset_len with total_epoch |
| 56 | self.config.buffer.explorer_input.tasksets[0].total_steps = None |
| 57 | self.config.buffer.explorer_input.tasksets[0].total_epochs = 3 |
| 58 | self.config.buffer.explorer_input.tasksets[0].index = 20 |
| 59 | reader = get_buffer_reader(self.config.buffer.explorer_input.tasksets[0]) |
| 60 | tasks = [] |
| 61 | while True: |
| 62 | try: |
| 63 | tasks.extend(await reader.read()) |
| 64 | except StopAsyncIteration: |
| 65 | break |
| 66 | self.assertEqual(len(tasks), 16 * 3 - 20) |
| 67 | |
| 68 | # test offset > dataset_len with total_steps |
| 69 | self.config.buffer.explorer_input.tasksets[0].total_steps = 10 |
| 70 | self.config.buffer.explorer_input.tasksets[0].index = 24 |
| 71 | reader = get_buffer_reader(self.config.buffer.explorer_input.tasksets[0]) |
| 72 | tasks = [] |
| 73 | while True: |
| 74 | try: |
nothing calls this directly
no test coverage detected