Function test_concurrent_cache_access

ci/test_all_lint_caches_stress.py:379–407

Test concurrent cache access doesn't break anything.

(test: CacheLintStressTest)

def test_concurrent_cache_access(test: CacheLintStressTest) -> None:
    """Test concurrent cache access doesn't break anything."""
    print("\n📋 TEST 8: Concurrent Cache Access")
    print("-" * 70)

    # Run 3 concurrent cache checks
    result_queue: multiprocessing.Queue[tuple[str, Any]] = multiprocessing.Queue()
    processes: list[multiprocessing.Process] = []

    for i in range(3):
        p = multiprocessing.Process(
            target=_concurrent_cache_check, args=("python", result_queue)
        )
        p.start()
        processes.append(p)

    # Wait for all
    for p in processes:
        p.join(timeout=15)

    # Check results
    results: list[tuple[str, Any]] = []
    while not result_queue.empty():
        results.append(result_queue.get())

    if len(results) >= 2:
        test.pass_test("Concurrent cache access completed without errors")
    else:
        test.fail_test("Concurrent cache access", f"Only {len(results)}/3 completed")


def main() -> int: