Function test_multiple_thread_stack

tests/integration/test_gather_stacks.py:108–166

(python, blocking, method, tmpdir)

@all_pystack_combinations()
def test_multiple_thread_stack(python, blocking, method, tmpdir):
    # GIVEN

    _, python_executable = python

    # WHEN

    with spawn_child_process(
        python_executable, TEST_MULTIPLE_THREADS_FILE, tmpdir
    ) as child_process:
        with xfail_on_expected_exceptions(method):
            threads = list(
                get_process_threads(
                    child_process.pid, stop_process=blocking, method=method
                )
            )

    # THEN

    assert len(threads) == 4
    main_thread = [
        thread
        for thread in threads
        if "threading" not in thread.first_frame.code.filename
    ][0]
    other_threads = [thread for thread in threads if thread != main_thread]

    frames = list(main_thread.frames)
    assert (len(frames)) == 4

    filenames = {frame.code.filename for frame in frames}
    assert filenames == {str(TEST_MULTIPLE_THREADS_FILE)}

    functions = [frame.code.scope for frame in frames]
    assert functions == ["<module>", "first_func", "second_func", "third_func"]

    *line_numbers, last_line = [frame.code.location.lineno for frame in frames]
    assert line_numbers == [42, 23, 27]
    assert last_line in {38, 39}

    assert not main_thread.native_frames

    for thread in other_threads:
        frames = [
            frame for frame in thread.frames if "threading" not in frame.code.filename
        ]
        assert (len(frames)) == 3

        filenames = {frame.code.filename for frame in frames}
        assert str(TEST_MULTIPLE_THREADS_FILE) in filenames

        functions = [frame.code.scope for frame in frames]
        assert functions == ["thread_func_1", "thread_func_2", "thread_func_3"]

        *line_numbers, last_line = [frame.code.location.lineno for frame in frames]
        assert line_numbers == [9, 13]
        assert last_line in {18, 19}