(lock_file: str)
| 240 | |
| 241 | @pytest.mark.timeout(_PROCESS_DEADLINE * 3) |
| 242 | def test_timeout_behavior(lock_file: str) -> None: |
| 243 | write_acquired = Event() |
| 244 | release_write = Event() |
| 245 | read_acquired = Event() |
| 246 | |
| 247 | writer = Process(target=acquire_lock, args=(lock_file, "write", write_acquired, release_write)) |
| 248 | reader = Process(target=acquire_lock, args=(lock_file, "read", read_acquired, None, 0.5, True)) |
| 249 | |
| 250 | with cleanup_processes([writer, reader]): |
| 251 | writer.start() |
| 252 | assert write_acquired.wait(timeout=_PROCESS_DEADLINE), "Write lock not acquired" |
| 253 | |
| 254 | start_time = time.time() |
| 255 | reader.start() |
| 256 | |
| 257 | assert not read_acquired.wait(timeout=1), "Read lock should not be acquired" |
| 258 | reader.join(timeout=_PROCESS_DEADLINE) |
| 259 | |
| 260 | elapsed = time.time() - start_time |
| 261 | assert 0.4 <= elapsed <= 10.0, f"Timeout was not respected: {elapsed}s" |
| 262 | |
| 263 | release_write.set() |
| 264 | writer.join(timeout=_PROCESS_DEADLINE) |
| 265 | |
| 266 | |
| 267 | @pytest.mark.timeout(_PROCESS_DEADLINE * 3) |
nothing calls this directly
no test coverage detected