test_multi.py 文件源码

def test_safe_terminate2(execmodel):
    if execmodel.backend != "threading":
        pytest.xfail("execution model %r does not support task count" %
                     execmodel.backend)
    import threading
    active = threading.active_count()
    l = []

    def term():
        return

    def kill():
        l.append(1)
    safe_terminate(execmodel, 3, [(term, kill)] * 10)
    assert len(l) == 0
    sleep(0.1)
    py.std.gc.collect()
    assert threading.active_count() == active