| 328 | |
| 329 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |
| 330 | def test_non_blocking(lock_type: type[BaseFileLock], tmp_path: Path) -> None: |
| 331 | # raises Timeout error when the lock cannot be acquired |
| 332 | lock_path = tmp_path / "a" |
| 333 | lock_1, lock_2 = lock_type(str(lock_path)), lock_type(str(lock_path)) |
| 334 | lock_3 = lock_type(str(lock_path), blocking=False) |
| 335 | lock_4 = lock_type(str(lock_path), timeout=0) |
| 336 | lock_5 = lock_type(str(lock_path), blocking=False, timeout=-1) |
| 337 | |
| 338 | # acquire lock 1 |
| 339 | lock_1.acquire() |
| 340 | assert lock_1.is_locked |
| 341 | assert not lock_2.is_locked |
| 342 | assert not lock_3.is_locked |
| 343 | assert not lock_4.is_locked |
| 344 | assert not lock_5.is_locked |
| 345 | |
| 346 | # try to acquire lock 2 |
| 347 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."): |
| 348 | lock_2.acquire(blocking=False) |
| 349 | assert not lock_2.is_locked |
| 350 | assert lock_1.is_locked |
| 351 | |
| 352 | # try to acquire pre-parametrized `blocking=False` lock 3 with `acquire` |
| 353 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."): |
| 354 | lock_3.acquire() |
| 355 | assert not lock_3.is_locked |
| 356 | assert lock_1.is_locked |
| 357 | |
| 358 | # try to acquire pre-parametrized `blocking=False` lock 3 with context manager |
| 359 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."), lock_3: |
| 360 | pass |
| 361 | assert not lock_3.is_locked |
| 362 | assert lock_1.is_locked |
| 363 | |
| 364 | # try to acquire pre-parametrized `timeout=0` lock 4 with `acquire` |
| 365 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."): |
| 366 | lock_4.acquire() |
| 367 | assert not lock_4.is_locked |
| 368 | assert lock_1.is_locked |
| 369 | |
| 370 | # try to acquire pre-parametrized `timeout=0` lock 4 with context manager |
| 371 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."), lock_4: |
| 372 | pass |
| 373 | assert not lock_4.is_locked |
| 374 | assert lock_1.is_locked |
| 375 | |
| 376 | # blocking precedence over timeout |
| 377 | # try to acquire pre-parametrized `timeout=-1,blocking=False` lock 5 with `acquire` |
| 378 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."): |
| 379 | lock_5.acquire() |
| 380 | assert not lock_5.is_locked |
| 381 | assert lock_1.is_locked |
| 382 | |
| 383 | # try to acquire pre-parametrized `timeout=-1,blocking=False` lock 5 with context manager |
| 384 | with pytest.raises(Timeout, match=r"The file lock '.*' could not be acquired."), lock_5: |
| 385 | pass |
| 386 | assert not lock_5.is_locked |
| 387 | assert lock_1.is_locked |