| 39 | self._per_batch_delay = per_batch_delay |
| 40 | |
| 41 | async def read_lines(self, timeout: float) -> AsyncIterator[str]: # noqa: ARG002 |
| 42 | if not self._batches: |
| 43 | # Simulate "nothing available within timeout" by sleeping the |
| 44 | # requested interval, like a real serial adapter would. |
| 45 | await asyncio.sleep(min(timeout, 0.01)) |
| 46 | return |
| 47 | batch = self._batches.pop(0) |
| 48 | for line in batch: |
| 49 | if self._per_batch_delay: |
| 50 | await asyncio.sleep(self._per_batch_delay) |
| 51 | yield line |
| 52 | |
| 53 | |
| 54 | def test_boot_signal_constants_are_populated() -> None: |