Test multiple processes checking cache simultaneously.
(results: StressTestResults)
| 183 | |
| 184 | |
| 185 | def test_concurrent_checks(results: StressTestResults) -> None: |
| 186 | """Test multiple processes checking cache simultaneously.""" |
| 187 | print("\n📋 TEST 3: Concurrent Access (Multiple Processes)") |
| 188 | print("-" * 70) |
| 189 | |
| 190 | with tempfile.TemporaryDirectory() as temp_dir: |
| 191 | temp_path = Path(temp_dir) |
| 192 | test_dir = temp_path / "test_files" |
| 193 | cache_dir = temp_path / ".cache" |
| 194 | |
| 195 | create_test_files(test_dir, 10) |
| 196 | root = str(test_dir) |
| 197 | |
| 198 | # Run 5 concurrent checks |
| 199 | result_queue: "multiprocessing.Queue[tuple[str, Any]]" = multiprocessing.Queue() |
| 200 | processes: list[multiprocessing.Process] = [] |
| 201 | |
| 202 | for i in range(5): |
| 203 | p = multiprocessing.Process( |
| 204 | target=_concurrent_check_worker, args=(cache_dir, root, result_queue) |
| 205 | ) |
| 206 | p.start() |
| 207 | processes.append(p) |
| 208 | |
| 209 | # Wait for all to complete |
| 210 | for p in processes: |
| 211 | p.join(timeout=30) |
| 212 | |
| 213 | # Check results |
| 214 | results_collected: list[tuple[str, Any]] = [] |
| 215 | while not result_queue.empty(): |
| 216 | item = result_queue.get() |
| 217 | if isinstance(item, tuple) and len(item) == 2: |
| 218 | status: str = item[0] |
| 219 | result: Any = item[1] |
| 220 | results_collected.append((status, result)) |
| 221 | |
| 222 | if len(results_collected) == 5: |
| 223 | results.pass_test("All 5 concurrent checks completed") |
| 224 | else: |
| 225 | results.fail_test( |
| 226 | "All concurrent checks completed", f"Only {len(results_collected)}/5" |
| 227 | ) |
| 228 | |
| 229 | # All should report cache miss (first run) |
| 230 | cache_misses = sum(1 for s, r in results_collected if s == "success" and r) |
| 231 | if cache_misses >= 4: |
| 232 | results.pass_test("Concurrent processes report consistent results") |
| 233 | else: |
| 234 | results.fail_test( |
| 235 | "Consistent concurrent results", |
| 236 | f"Only {cache_misses}/5 reported cache miss", |
| 237 | ) |
| 238 | |
| 239 | |
| 240 | def test_cross_process_pending_fingerprint(results: StressTestResults) -> None: |