()
| 127 | holder.start() |
| 128 | |
| 129 | async def measure() -> float: |
| 130 | waiter = AsyncFileLock(path) |
| 131 | task = asyncio.ensure_future(waiter.acquire(poll_interval=0.01)) |
| 132 | await asyncio.sleep(0.02) |
| 133 | started = time.perf_counter() |
| 134 | task.cancel() |
| 135 | with suppress(asyncio.CancelledError): |
| 136 | await task |
| 137 | return time.perf_counter() - started |
| 138 | |
| 139 | try: |
| 140 | acquired.wait(timeout=5) |
no test coverage detected