Test that timeout parameter works correctly in multi-process environment.
(lock_file: str)
| 311 | |
| 312 | @pytest.mark.timeout(10) |
| 313 | def test_timeout_behavior(lock_file: str) -> None: |
| 314 | """Test that timeout parameter works correctly in multi-process environment.""" |
| 315 | write_acquired = Event() |
| 316 | release_write = Event() |
| 317 | read_acquired = Event() |
| 318 | |
| 319 | writer = Process(target=acquire_lock, args=(lock_file, "write", write_acquired, release_write)) |
| 320 | reader = Process(target=acquire_lock, args=(lock_file, "read", read_acquired, None, 0.5, True)) |
| 321 | |
| 322 | with cleanup_processes([writer, reader]): |
| 323 | writer.start() |
| 324 | assert write_acquired.wait(timeout=2), "Write lock not acquired" |
| 325 | |
| 326 | start_time = time.time() |
| 327 | reader.start() |
| 328 | |
| 329 | assert not read_acquired.wait(timeout=1), "Read lock should not be acquired" |
| 330 | reader.join(timeout=5) |
| 331 | |
| 332 | elapsed = time.time() - start_time |
| 333 | assert 0.4 <= elapsed <= 10.0, f"Timeout was not respected: {elapsed}s" |
| 334 | |
| 335 | release_write.set() |
| 336 | writer.join(timeout=2) |
| 337 | |
| 338 | |
| 339 | @pytest.mark.timeout(10) |
nothing calls this directly
no test coverage detected
searching dependent graphs…