| 244 | |
| 245 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |
| 246 | def test_threaded_shared_lock_obj(lock_type: type[BaseFileLock], tmp_path: Path) -> None: |
| 247 | if sys.platform == "win32" and lock_type.__name__ == "SoftFileLock": |
| 248 | pytest.skip( |
| 249 | "SoftFileLock uses file-existence locking — on Windows, unlink can silently fail under heavy " |
| 250 | "thread contention (EACCES from antivirus/indexer), orphaning the lock file with no recovery path" |
| 251 | ) |
| 252 | |
| 253 | # Runs 100 threads, which need the filelock. The lock must be acquired if at least one thread required it and |
| 254 | # released, as soon as all threads stopped. |
| 255 | lock_path = tmp_path / "a" |
| 256 | lock = lock_type(str(lock_path)) |
| 257 | |
| 258 | def thread_work() -> None: |
| 259 | for _ in range(100): |
| 260 | with lock: |
| 261 | assert lock.is_locked |
| 262 | |
| 263 | threads = [ExThread(target=thread_work, name=f"t{i}") for i in range(100)] |
| 264 | for thread in threads: |
| 265 | thread.start() |
| 266 | for thread in threads: |
| 267 | thread.join() |
| 268 | |
| 269 | assert not lock.is_locked |
| 270 | |
| 271 | |
| 272 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |