(lock_file: str, mode: Literal["read", "write"])
| 61 | @pytest.mark.parametrize("mode", [pytest.param("read", id="read"), pytest.param("write", id="write")]) |
| 62 | @pytest.mark.asyncio |
| 63 | async def test_reentrant(lock_file: str, mode: Literal["read", "write"]) -> None: |
| 64 | lock = AsyncReadWriteLock(lock_file, is_singleton=False) |
| 65 | acquire = lock.acquire_read if mode == "read" else lock.acquire_write |
| 66 | await acquire() |
| 67 | await acquire() |
| 68 | await lock.release() |
| 69 | assert_mode_held(lock_file, mode) |
| 70 | await lock.release() |
| 71 | with pytest.raises(RuntimeError, match="not held"): |
| 72 | await lock.release() |
| 73 | await lock.close() |
| 74 | |
| 75 | |
| 76 | @pytest.mark.parametrize( |
nothing calls this directly
no test coverage detected