Test Manager shared between parent and child process.
()
| 28 | |
| 29 | |
| 30 | def test_manager_with_process(): |
| 31 | """Test Manager shared between parent and child process.""" |
| 32 | print("\n=== Test 2: Manager with forked child ===") |
| 33 | ctx = multiprocessing.get_context("fork") |
| 34 | manager = ctx.Manager() |
| 35 | try: |
| 36 | result = manager.Value("i", 0) |
| 37 | ev = manager.Event() |
| 38 | |
| 39 | def child_fn(): |
| 40 | try: |
| 41 | ev.set() |
| 42 | result.value = 42 |
| 43 | except Exception as e: |
| 44 | print(f" CHILD ERROR: {e}", file=sys.stderr) |
| 45 | traceback.print_exc() |
| 46 | sys.exit(1) |
| 47 | |
| 48 | print(f" Starting child process...") |
| 49 | process = ctx.Process(target=child_fn) |
| 50 | process.start() |
| 51 | print(f" Waiting for child (pid={process.pid})...") |
| 52 | process.join(timeout=10) |
| 53 | |
| 54 | if process.exitcode != 0: |
| 55 | print(f" FAIL: child exited with code {process.exitcode}") |
| 56 | return False |
| 57 | |
| 58 | print(f" Child done. result={result.value}, event={ev.is_set()}") |
| 59 | assert result.value == 42 |
| 60 | assert ev.is_set() |
| 61 | print(" PASS") |
| 62 | return True |
| 63 | finally: |
| 64 | manager.shutdown() |
| 65 | |
| 66 | |
| 67 | def test_manager_server_alive_after_fork(): |