| 271 | |
| 272 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |
| 273 | def test_threaded_lock_different_lock_obj(lock_type: type[BaseFileLock], tmp_path: Path) -> None: |
| 274 | if sys.platform == "win32" and (hasattr(sys, "pypy_version_info") or lock_type.__name__ == "SoftFileLock"): |
| 275 | pytest.skip("SoftFileLock on Windows has race conditions under heavy threading") |
| 276 | |
| 277 | # Runs multiple threads, which acquire the same lock file with a different FileLock object. When thread group 1 |
| 278 | # acquired the lock, thread group 2 must not hold their lock. |
| 279 | def t_1() -> None: |
| 280 | for _ in range(1000): |
| 281 | with lock_1: |
| 282 | assert lock_1.is_locked |
| 283 | assert not lock_2.is_locked |
| 284 | |
| 285 | def t_2() -> None: |
| 286 | for _ in range(1000): |
| 287 | with lock_2: |
| 288 | assert not lock_1.is_locked |
| 289 | assert lock_2.is_locked |
| 290 | |
| 291 | lock_path = tmp_path / "a" |
| 292 | lock_1, lock_2 = lock_type(str(lock_path)), lock_type(str(lock_path)) |
| 293 | threads = [(ExThread(t_1, f"t1_{i}"), ExThread(t_2, f"t2_{i}")) for i in range(10)] |
| 294 | |
| 295 | for thread_1, thread_2 in threads: |
| 296 | thread_1.start() |
| 297 | thread_2.start() |
| 298 | for thread_1, thread_2 in threads: |
| 299 | thread_1.join() |
| 300 | thread_2.join() |
| 301 | |
| 302 | assert not lock_1.is_locked |
| 303 | assert not lock_2.is_locked |
| 304 | |
| 305 | |
| 306 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |