test_async.py 文件源码

def test_run_async_exit_code(self):

        def run_with_exit_code_0(process_idx):
            sys.exit(0)

        def run_with_exit_code_11(process_idx):
            os.kill(os.getpid(), signal.SIGSEGV)

        with warnings.catch_warnings(record=True) as w:
            async.run_async(4, run_with_exit_code_0)
            # There should be no warnings
            assert len(w) == 0

        with warnings.catch_warnings(record=True) as w:
            async.run_async(4, run_with_exit_code_11)
            # There should be 4 warnings
            assert len(w) == 4