Worker function for concurrent cache check (must be module-level for Windows pickling).
(
cache_type: str, result_queue: multiprocessing.Queue[tuple[str, Any]]
)
| 356 | |
| 357 | |
| 358 | def _concurrent_cache_check( |
| 359 | cache_type: str, result_queue: multiprocessing.Queue[tuple[str, Any]] |
| 360 | ) -> None: |
| 361 | """Worker function for concurrent cache check (must be module-level for Windows pickling).""" |
| 362 | try: |
| 363 | result = subprocess.run( |
| 364 | ["uv", "run", "python", "ci/python_lint_cache.py", "check"], |
| 365 | capture_output=True, |
| 366 | text=True, |
| 367 | encoding="utf-8", |
| 368 | errors="replace", |
| 369 | timeout=10, |
| 370 | ) |
| 371 | result_queue.put(("success", result.returncode)) |
| 372 | except KeyboardInterrupt as ki: |
| 373 | handle_keyboard_interrupt(ki) |
| 374 | raise |
| 375 | except Exception as e: |
| 376 | result_queue.put(("error", str(e))) |
| 377 | |
| 378 | |
| 379 | def test_concurrent_cache_access(test: CacheLintStressTest) -> None: |
nothing calls this directly
no test coverage detected