|
17 | 17 | import pytest |
18 | 18 | from capabilities import CAPABILITIES |
19 | 19 |
|
20 | | -from filelock import Timeout |
| 20 | +from filelock import AsyncSoftReadWriteLock, Timeout |
21 | 21 | from filelock import _util as util_mod |
22 | 22 | from filelock._soft_rw import SoftReadWriteLock |
23 | 23 | from filelock._soft_rw import _sync as sync_mod |
@@ -53,6 +53,45 @@ def lock_file(tmp_path: Path) -> str: |
53 | 53 | return str(tmp_path / "test.lock") |
54 | 54 |
|
55 | 55 |
|
| 56 | +@pytest.mark.parametrize( |
| 57 | + "lock_type", |
| 58 | + [pytest.param(SoftReadWriteLock, id="sync"), pytest.param(AsyncSoftReadWriteLock, id="async")], |
| 59 | +) |
| 60 | +@pytest.mark.parametrize("cached", [pytest.param(False, id="new"), pytest.param(True, id="cached")]) |
| 61 | +@pytest.mark.parametrize( |
| 62 | + ("settings", "message"), |
| 63 | + [ |
| 64 | + pytest.param({name: value}, name, id=f"{name}-{label}") |
| 65 | + for name in ("heartbeat_interval", "stale_threshold", "poll_interval") |
| 66 | + for label, value in (("nan", float("nan")), ("infinity", float("inf")), ("negative-infinity", float("-inf"))) |
| 67 | + ] |
| 68 | + + [ |
| 69 | + pytest.param({"heartbeat_interval": sys.float_info.max}, "stale_threshold", id="default-overflow"), |
| 70 | + ], |
| 71 | +) |
| 72 | +def test_rejects_invalid_intervals( |
| 73 | + tmp_path: Path, |
| 74 | + lock_type: type[SoftReadWriteLock | AsyncSoftReadWriteLock], |
| 75 | + cached: bool, |
| 76 | + settings: dict[str, float], |
| 77 | + message: str, |
| 78 | +) -> None: |
| 79 | + path: Final = tmp_path / "timing.lock" |
| 80 | + # Keep the weakly cached instance alive through the second construction. |
| 81 | + existing: Final = SoftReadWriteLock(path) if cached else None |
| 82 | + try: |
| 83 | + with pytest.raises(ValueError, match=rf"{message} must .*finite"): |
| 84 | + lock_type( |
| 85 | + path, |
| 86 | + heartbeat_interval=settings.get("heartbeat_interval", 30), |
| 87 | + stale_threshold=settings.get("stale_threshold"), |
| 88 | + poll_interval=settings.get("poll_interval", 0.25), |
| 89 | + ) |
| 90 | + finally: |
| 91 | + if existing is not None: |
| 92 | + existing.close() |
| 93 | + |
| 94 | + |
56 | 95 | def test_rejects_non_positive_heartbeat_interval(lock_file: str) -> None: |
57 | 96 | with pytest.raises(ValueError, match="heartbeat_interval must be positive"): |
58 | 97 | SoftReadWriteLock(lock_file, heartbeat_interval=0, is_singleton=False) |
|
0 commit comments