Check if there are no children processes started by the test after it ended. Be sure to call `capture_processes` in the test.
()
| 40 | |
| 41 | |
| 42 | def teardown_function(): |
| 43 | """Check if there are no children processes started by the test after it ended. |
| 44 | |
| 45 | Be sure to call `capture_processes` in the test. |
| 46 | """ |
| 47 | assert len(pool_processes), "No processes where tracked - did the test call capture_processes?" |
| 48 | pools_not_collected = [pool_ref() is not None for pool_ref in pools] |
| 49 | current_process = psutil.Process() |
| 50 | children_pids = [process.pid for process in current_process.children()] |
| 51 | left = set(pool_processes).intersection(children_pids) |
| 52 | assert len(left) == 0, ( |
| 53 | f"Pipeline-started processes left after test is finished, pids alive: {left},\n" |
| 54 | f"pids started during tests: {pool_processes}.\n" |
| 55 | f"Pools not collected: {sum(pools_not_collected)}" |
| 56 | ) |
| 57 | alive_threads = [thread.is_alive() for thread in pool_threads] |
| 58 | assert sum(alive_threads) == 0, ( |
| 59 | "Some pool related threads are left after the test finished. " |
| 60 | "Started in test suite: {}, still active: {}. " |
| 61 | "Active threads map in the order of creation {}".format( |
| 62 | len(pool_threads), sum(alive_threads), alive_threads |
| 63 | ) |
| 64 | ) |
| 65 | |
| 66 | |
| 67 | def check_shm_for_dali(msg): |