Function test_threading

tests/test_iostream.py:313–329

()

@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_threading():
    with m.ostream_redirect(stdout=True, stderr=False):
        # start some threads
        threads = [m.TestThread() for _j in range(20)]

        # give the threads some time to fail
        threads[0].sleep()

        # stop all the threads
        for t in threads:
            t.stop()

        for t in threads:
            t.join()

        # if a thread segfaults, we don't get here
        assert True