Function capture_test_stack

Lib/test/test_asyncio/test_graph.py:11–45

(*, fut=None, depth=1)

def capture_test_stack(*, fut=None, depth=1):

    def walk(s):
        ret = [
            (f"T<{n}>" if '-' not in (n := s.future.get_name()) else 'T<anon>')
                if isinstance(s.future, asyncio.Task) else 'F'
        ]

        ret.append(
            [
                (
                    f"s {entry.frame.f_code.co_name}"
                        if entry.frame.f_generator is None else
                        (
                            f"a {entry.frame.f_generator.cr_code.co_name}"
                            if hasattr(entry.frame.f_generator, 'cr_code') else
                            f"ag {entry.frame.f_generator.ag_code.co_name}"
                        )
                ) for entry in s.call_stack
            ]
        )

        ret.append(
            sorted([
                walk(ab) for ab in s.awaited_by
            ], key=lambda entry: entry[0])
        )

        return ret

    buf = io.StringIO()
    asyncio.print_call_graph(fut, file=buf, depth=depth+1)

    stack = asyncio.capture_call_graph(fut, depth=depth)
    return walk(stack), buf.getvalue()


class CallStackTestBase: