| import unittest | |
| from asyncio import tools | |
| import _remote_debugging | |
| def LocationInfo(lineno, end_lineno=None, col_offset=None, end_col_offset=None): | |
| return _remote_debugging.LocationInfo((lineno, end_lineno, col_offset, end_col_offset)) | |
| def FrameInfo(funcname, filename, location, opcode=None): | |
| return _remote_debugging.FrameInfo((filename, location, funcname, opcode)) | |
| def CoroInfo(call_stack, task_name): | |
| return _remote_debugging.CoroInfo((call_stack, task_name)) | |
| def TaskInfo(task_id, task_name, coroutine_stack, awaited_by): | |
| return _remote_debugging.TaskInfo((task_id, task_name, coroutine_stack, awaited_by)) | |
| def AwaitedInfo(thread_id, awaited_by): | |
| return _remote_debugging.AwaitedInfo((thread_id, awaited_by)) | |
| # mock output of get_all_awaited_by function. | |
| TEST_INPUTS_TREE = [ | |
| [ | |
| # test case containing a task called timer being awaited in two | |
| # different subtasks part of a TaskGroup (root1 and root2) which call | |
| # awaiter functions. | |
| ( | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="timer", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiter3", "/path/to/app.py", LocationInfo(130)), | |
| FrameInfo("awaiter2", "/path/to/app.py", LocationInfo(120)), | |
| FrameInfo("awaiter", "/path/to/app.py", LocationInfo(110)) | |
| ], | |
| task_name=4 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiterB3", "/path/to/app.py", LocationInfo(190)), | |
| FrameInfo("awaiterB2", "/path/to/app.py", LocationInfo(180)), | |
| FrameInfo("awaiterB", "/path/to/app.py", LocationInfo(170)) | |
| ], | |
| task_name=5 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiterB3", "/path/to/app.py", LocationInfo(190)), | |
| FrameInfo("awaiterB2", "/path/to/app.py", LocationInfo(180)), | |
| FrameInfo("awaiterB", "/path/to/app.py", LocationInfo(170)) | |
| ], | |
| task_name=6 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiter3", "/path/to/app.py", LocationInfo(130)), | |
| FrameInfo("awaiter2", "/path/to/app.py", LocationInfo(120)), | |
| FrameInfo("awaiter", "/path/to/app.py", LocationInfo(110)) | |
| ], | |
| task_name=7 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=8, | |
| task_name="root1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("main", "", LocationInfo(0)) | |
| ], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=9, | |
| task_name="root2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("main", "", LocationInfo(0)) | |
| ], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="child1_1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=8 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="child2_1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=8 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=7, | |
| task_name="child1_2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=9 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=5, | |
| task_name="child2_2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=9 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ), | |
| ( | |
| [ | |
| [ | |
| "└── (T) Task-1", | |
| " └── main", | |
| " └── __aexit__", | |
| " └── _aexit", | |
| " ├── (T) root1", | |
| " │ └── bloch", | |
| " │ └── blocho_caller", | |
| " │ └── __aexit__", | |
| " │ └── _aexit", | |
| " │ ├── (T) child1_1", | |
| " │ │ └── awaiter /path/to/app.py:110", | |
| " │ │ └── awaiter2 /path/to/app.py:120", | |
| " │ │ └── awaiter3 /path/to/app.py:130", | |
| " │ │ └── (T) timer", | |
| " │ └── (T) child2_1", | |
| " │ └── awaiterB /path/to/app.py:170", | |
| " │ └── awaiterB2 /path/to/app.py:180", | |
| " │ └── awaiterB3 /path/to/app.py:190", | |
| " │ └── (T) timer", | |
| " └── (T) root2", | |
| " └── bloch", | |
| " └── blocho_caller", | |
| " └── __aexit__", | |
| " └── _aexit", | |
| " ├── (T) child1_2", | |
| " │ └── awaiter /path/to/app.py:110", | |
| " │ └── awaiter2 /path/to/app.py:120", | |
| " │ └── awaiter3 /path/to/app.py:130", | |
| " │ └── (T) timer", | |
| " └── (T) child2_2", | |
| " └── awaiterB /path/to/app.py:170", | |
| " └── awaiterB2 /path/to/app.py:180", | |
| " └── awaiterB3 /path/to/app.py:190", | |
| " └── (T) timer", | |
| ] | |
| ] | |
| ), | |
| ], | |
| [ | |
| # test case containing two roots | |
| ( | |
| AwaitedInfo( | |
| thread_id=9, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=5, | |
| task_name="Task-5", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="Task-6", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main2", "", LocationInfo(0))], | |
| task_name=5 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=7, | |
| task_name="Task-7", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main2", "", LocationInfo(0))], | |
| task_name=5 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=8, | |
| task_name="Task-8", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main2", "", LocationInfo(0))], | |
| task_name=5 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo( | |
| thread_id=10, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="Task-4", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=11, awaited_by=[]), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ), | |
| ( | |
| [ | |
| [ | |
| "└── (T) Task-5", | |
| " └── main2", | |
| " ├── (T) Task-6", | |
| " ├── (T) Task-7", | |
| " └── (T) Task-8", | |
| ], | |
| [ | |
| "└── (T) Task-1", | |
| " └── main", | |
| " ├── (T) Task-2", | |
| " ├── (T) Task-3", | |
| " └── (T) Task-4", | |
| ], | |
| ] | |
| ), | |
| ], | |
| [ | |
| # test case containing two roots, one of them without subtasks | |
| ( | |
| [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-5", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo( | |
| thread_id=3, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=4, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=5, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=4 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=4 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=7, | |
| task_name="Task-4", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=4 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=8, awaited_by=[]), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ] | |
| ), | |
| ( | |
| [ | |
| ["└── (T) Task-5"], | |
| [ | |
| "└── (T) Task-1", | |
| " └── main", | |
| " ├── (T) Task-2", | |
| " ├── (T) Task-3", | |
| " └── (T) Task-4", | |
| ], | |
| ] | |
| ), | |
| ], | |
| ] | |
| TEST_INPUTS_CYCLES_TREE = [ | |
| [ | |
| # this test case contains a cycle: two tasks awaiting each other. | |
| ( | |
| [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="a", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("awaiter2", "", LocationInfo(0))], | |
| task_name=4 | |
| ), | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="b", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("awaiter", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ] | |
| ), | |
| ([[4, 3, 4]]), | |
| ], | |
| [ | |
| # this test case contains two cycles | |
| ( | |
| [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="A", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_b", "", LocationInfo(0)) | |
| ], | |
| task_name=4 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="B", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_c", "", LocationInfo(0)) | |
| ], | |
| task_name=5 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_a", "", LocationInfo(0)) | |
| ], | |
| task_name=3 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=5, | |
| task_name="C", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)) | |
| ], | |
| task_name=6 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_b", "", LocationInfo(0)) | |
| ], | |
| task_name=4 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ] | |
| ), | |
| ([[4, 3, 4], [4, 6, 5, 4]]), | |
| ], | |
| ] | |
| TEST_INPUTS_TABLE = [ | |
| [ | |
| # test case containing a task called timer being awaited in two | |
| # different subtasks part of a TaskGroup (root1 and root2) which call | |
| # awaiter functions. | |
| ( | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="timer", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiter3", "", LocationInfo(0)), | |
| FrameInfo("awaiter2", "", LocationInfo(0)), | |
| FrameInfo("awaiter", "", LocationInfo(0)) | |
| ], | |
| task_name=4 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiter1_3", "", LocationInfo(0)), | |
| FrameInfo("awaiter1_2", "", LocationInfo(0)), | |
| FrameInfo("awaiter1", "", LocationInfo(0)) | |
| ], | |
| task_name=5 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiter1_3", "", LocationInfo(0)), | |
| FrameInfo("awaiter1_2", "", LocationInfo(0)), | |
| FrameInfo("awaiter1", "", LocationInfo(0)) | |
| ], | |
| task_name=6 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("awaiter3", "", LocationInfo(0)), | |
| FrameInfo("awaiter2", "", LocationInfo(0)), | |
| FrameInfo("awaiter", "", LocationInfo(0)) | |
| ], | |
| task_name=7 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=8, | |
| task_name="root1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("main", "", LocationInfo(0)) | |
| ], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=9, | |
| task_name="root2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("main", "", LocationInfo(0)) | |
| ], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="child1_1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=8 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="child2_1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=8 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=7, | |
| task_name="child1_2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=9 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=5, | |
| task_name="child2_2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("_aexit", "", LocationInfo(0)), | |
| FrameInfo("__aexit__", "", LocationInfo(0)), | |
| FrameInfo("blocho_caller", "", LocationInfo(0)), | |
| FrameInfo("bloch", "", LocationInfo(0)) | |
| ], | |
| task_name=9 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ), | |
| ( | |
| [ | |
| [1, "0x2", "Task-1", "", "", "", "0x0"], | |
| [ | |
| 1, | |
| "0x3", | |
| "timer", | |
| "", | |
| "awaiter3 -> awaiter2 -> awaiter", | |
| "child1_1", | |
| "0x4", | |
| ], | |
| [ | |
| 1, | |
| "0x3", | |
| "timer", | |
| "", | |
| "awaiter1_3 -> awaiter1_2 -> awaiter1", | |
| "child2_2", | |
| "0x5", | |
| ], | |
| [ | |
| 1, | |
| "0x3", | |
| "timer", | |
| "", | |
| "awaiter1_3 -> awaiter1_2 -> awaiter1", | |
| "child2_1", | |
| "0x6", | |
| ], | |
| [ | |
| 1, | |
| "0x3", | |
| "timer", | |
| "", | |
| "awaiter3 -> awaiter2 -> awaiter", | |
| "child1_2", | |
| "0x7", | |
| ], | |
| [ | |
| 1, | |
| "0x8", | |
| "root1", | |
| "", | |
| "_aexit -> __aexit__ -> main", | |
| "Task-1", | |
| "0x2", | |
| ], | |
| [ | |
| 1, | |
| "0x9", | |
| "root2", | |
| "", | |
| "_aexit -> __aexit__ -> main", | |
| "Task-1", | |
| "0x2", | |
| ], | |
| [ | |
| 1, | |
| "0x4", | |
| "child1_1", | |
| "", | |
| "_aexit -> __aexit__ -> blocho_caller -> bloch", | |
| "root1", | |
| "0x8", | |
| ], | |
| [ | |
| 1, | |
| "0x6", | |
| "child2_1", | |
| "", | |
| "_aexit -> __aexit__ -> blocho_caller -> bloch", | |
| "root1", | |
| "0x8", | |
| ], | |
| [ | |
| 1, | |
| "0x7", | |
| "child1_2", | |
| "", | |
| "_aexit -> __aexit__ -> blocho_caller -> bloch", | |
| "root2", | |
| "0x9", | |
| ], | |
| [ | |
| 1, | |
| "0x5", | |
| "child2_2", | |
| "", | |
| "_aexit -> __aexit__ -> blocho_caller -> bloch", | |
| "root2", | |
| "0x9", | |
| ], | |
| ] | |
| ), | |
| ], | |
| [ | |
| # test case containing two roots | |
| ( | |
| AwaitedInfo( | |
| thread_id=9, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=5, | |
| task_name="Task-5", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="Task-6", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main2", "", LocationInfo(0))], | |
| task_name=5 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=7, | |
| task_name="Task-7", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main2", "", LocationInfo(0))], | |
| task_name=5 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=8, | |
| task_name="Task-8", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main2", "", LocationInfo(0))], | |
| task_name=5 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo( | |
| thread_id=10, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="Task-4", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=11, awaited_by=[]), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ), | |
| ( | |
| [ | |
| [9, "0x5", "Task-5", "", "", "", "0x0"], | |
| [9, "0x6", "Task-6", "", "main2", "Task-5", "0x5"], | |
| [9, "0x7", "Task-7", "", "main2", "Task-5", "0x5"], | |
| [9, "0x8", "Task-8", "", "main2", "Task-5", "0x5"], | |
| [10, "0x1", "Task-1", "", "", "", "0x0"], | |
| [10, "0x2", "Task-2", "", "main", "Task-1", "0x1"], | |
| [10, "0x3", "Task-3", "", "main", "Task-1", "0x1"], | |
| [10, "0x4", "Task-4", "", "main", "Task-1", "0x1"], | |
| ] | |
| ), | |
| ], | |
| [ | |
| # test case containing two roots, one of them without subtasks | |
| ( | |
| [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-5", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo( | |
| thread_id=3, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=4, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=5, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=4 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=4 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=7, | |
| task_name="Task-4", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=4 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=8, awaited_by=[]), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ] | |
| ), | |
| ( | |
| [ | |
| [1, "0x2", "Task-5", "", "", "", "0x0"], | |
| [3, "0x4", "Task-1", "", "", "", "0x0"], | |
| [3, "0x5", "Task-2", "", "main", "Task-1", "0x4"], | |
| [3, "0x6", "Task-3", "", "main", "Task-1", "0x4"], | |
| [3, "0x7", "Task-4", "", "main", "Task-1", "0x4"], | |
| ] | |
| ), | |
| ], | |
| # CASES WITH CYCLES | |
| [ | |
| # this test case contains a cycle: two tasks awaiting each other. | |
| ( | |
| [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="a", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("awaiter2", "", LocationInfo(0))], | |
| task_name=4 | |
| ), | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="b", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("awaiter", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ] | |
| ), | |
| ( | |
| [ | |
| [1, "0x2", "Task-1", "", "", "", "0x0"], | |
| [1, "0x3", "a", "", "awaiter2", "b", "0x4"], | |
| [1, "0x3", "a", "", "main", "Task-1", "0x2"], | |
| [1, "0x4", "b", "", "awaiter", "a", "0x3"], | |
| ] | |
| ), | |
| ], | |
| [ | |
| # this test case contains two cycles | |
| ( | |
| [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="A", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_b", "", LocationInfo(0)) | |
| ], | |
| task_name=4 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="B", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_c", "", LocationInfo(0)) | |
| ], | |
| task_name=5 | |
| ), | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_a", "", LocationInfo(0)) | |
| ], | |
| task_name=3 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=5, | |
| task_name="C", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)) | |
| ], | |
| task_name=6 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=6, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("nested", "", LocationInfo(0)), | |
| FrameInfo("task_b", "", LocationInfo(0)) | |
| ], | |
| task_name=4 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ), | |
| AwaitedInfo(thread_id=0, awaited_by=[]) | |
| ] | |
| ), | |
| ( | |
| [ | |
| [1, "0x2", "Task-1", "", "", "", "0x0"], | |
| [ | |
| 1, | |
| "0x3", | |
| "A", | |
| "", | |
| "nested -> nested -> task_b", | |
| "B", | |
| "0x4", | |
| ], | |
| [ | |
| 1, | |
| "0x4", | |
| "B", | |
| "", | |
| "nested -> nested -> task_c", | |
| "C", | |
| "0x5", | |
| ], | |
| [ | |
| 1, | |
| "0x4", | |
| "B", | |
| "", | |
| "nested -> nested -> task_a", | |
| "A", | |
| "0x3", | |
| ], | |
| [ | |
| 1, | |
| "0x5", | |
| "C", | |
| "", | |
| "nested -> nested", | |
| "Task-2", | |
| "0x6", | |
| ], | |
| [ | |
| 1, | |
| "0x6", | |
| "Task-2", | |
| "", | |
| "nested -> nested -> task_b", | |
| "B", | |
| "0x4", | |
| ], | |
| ] | |
| ), | |
| ], | |
| ] | |
| class TestAsyncioToolsTree(unittest.TestCase): | |
| def test_asyncio_utils(self): | |
| for input_, tree in TEST_INPUTS_TREE: | |
| with self.subTest(input_): | |
| result = tools.build_async_tree(input_) | |
| self.assertEqual(result, tree) | |
| def test_asyncio_utils_cycles(self): | |
| for input_, cycles in TEST_INPUTS_CYCLES_TREE: | |
| with self.subTest(input_): | |
| try: | |
| tools.build_async_tree(input_) | |
| except tools.CycleFoundException as e: | |
| self.assertEqual(e.cycles, cycles) | |
| class TestAsyncioToolsTable(unittest.TestCase): | |
| def test_asyncio_utils(self): | |
| for input_, table in TEST_INPUTS_TABLE: | |
| with self.subTest(input_): | |
| result = tools.build_task_table(input_) | |
| self.assertEqual(result, table) | |
| class TestAsyncioToolsBasic(unittest.TestCase): | |
| def test_empty_input_tree(self): | |
| """Test build_async_tree with empty input.""" | |
| result = [] | |
| expected_output = [] | |
| self.assertEqual(tools.build_async_tree(result), expected_output) | |
| def test_empty_input_table(self): | |
| """Test build_task_table with empty input.""" | |
| result = [] | |
| expected_output = [] | |
| self.assertEqual(tools.build_task_table(result), expected_output) | |
| def test_only_independent_tasks_tree(self): | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=10, | |
| task_name="taskA", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=11, | |
| task_name="taskB", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| expected = [["└── (T) taskA"], ["└── (T) taskB"]] | |
| result = tools.build_async_tree(input_) | |
| self.assertEqual(sorted(result), sorted(expected)) | |
| def test_only_independent_tasks_table(self): | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=10, | |
| task_name="taskA", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=11, | |
| task_name="taskB", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| self.assertEqual( | |
| tools.build_task_table(input_), | |
| [[1, '0xa', 'taskA', '', '', '', '0x0'], [1, '0xb', 'taskB', '', '', '', '0x0']] | |
| ) | |
| def test_single_task_tree(self): | |
| """Test build_async_tree with a single task and no awaits.""" | |
| result = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| expected_output = [ | |
| [ | |
| "└── (T) Task-1", | |
| ] | |
| ] | |
| self.assertEqual(tools.build_async_tree(result), expected_output) | |
| def test_single_task_table(self): | |
| """Test build_task_table with a single task and no awaits.""" | |
| result = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| expected_output = [[1, '0x2', 'Task-1', '', '', '', '0x0']] | |
| self.assertEqual(tools.build_task_table(result), expected_output) | |
| def test_cycle_detection(self): | |
| """Test build_async_tree raises CycleFoundException for cyclic input.""" | |
| result = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| with self.assertRaises(tools.CycleFoundException) as context: | |
| tools.build_async_tree(result) | |
| self.assertEqual(context.exception.cycles, [[3, 2, 3]]) | |
| def test_complex_tree(self): | |
| """Test build_async_tree with a more complex tree structure.""" | |
| result = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| expected_output = [ | |
| [ | |
| "└── (T) Task-1", | |
| " └── main", | |
| " └── (T) Task-2", | |
| " └── main", | |
| " └── (T) Task-3", | |
| ] | |
| ] | |
| self.assertEqual(tools.build_async_tree(result), expected_output) | |
| def test_complex_table(self): | |
| """Test build_task_table with a more complex tree structure.""" | |
| result = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=4, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("main", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| expected_output = [ | |
| [1, '0x2', 'Task-1', '', '', '', '0x0'], | |
| [1, '0x3', 'Task-2', '', 'main', 'Task-1', '0x2'], | |
| [1, '0x4', 'Task-3', '', 'main', 'Task-2', '0x3'] | |
| ] | |
| self.assertEqual(tools.build_task_table(result), expected_output) | |
| def test_deep_coroutine_chain(self): | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=10, | |
| task_name="leaf", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("c1", "", LocationInfo(0)), | |
| FrameInfo("c2", "", LocationInfo(0)), | |
| FrameInfo("c3", "", LocationInfo(0)), | |
| FrameInfo("c4", "", LocationInfo(0)), | |
| FrameInfo("c5", "", LocationInfo(0)) | |
| ], | |
| task_name=11 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=11, | |
| task_name="root", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| expected = [ | |
| [ | |
| "└── (T) root", | |
| " └── c5", | |
| " └── c4", | |
| " └── c3", | |
| " └── c2", | |
| " └── c1", | |
| " └── (T) leaf", | |
| ] | |
| ] | |
| result = tools.build_async_tree(input_) | |
| self.assertEqual(result, expected) | |
| def test_multiple_cycles_same_node(self): | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-A", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("call1", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-B", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("call2", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-C", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("call3", "", LocationInfo(0))], | |
| task_name=1 | |
| ), | |
| CoroInfo( | |
| call_stack=[FrameInfo("call4", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| with self.assertRaises(tools.CycleFoundException) as ctx: | |
| tools.build_async_tree(input_) | |
| cycles = ctx.exception.cycles | |
| self.assertTrue(any(set(c) == {1, 2, 3} for c in cycles)) | |
| def test_table_output_format(self): | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-A", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("foo", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-B", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| table = tools.build_task_table(input_) | |
| for row in table: | |
| self.assertEqual(len(row), 7) | |
| self.assertIsInstance(row[0], int) # thread ID | |
| self.assertTrue( | |
| isinstance(row[1], str) and row[1].startswith("0x") | |
| ) # hex task ID | |
| self.assertIsInstance(row[2], str) # task name | |
| self.assertIsInstance(row[3], str) # coroutine stack | |
| self.assertIsInstance(row[4], str) # coroutine chain | |
| self.assertIsInstance(row[5], str) # awaiter name | |
| self.assertTrue( | |
| isinstance(row[6], str) and row[6].startswith("0x") | |
| ) # hex awaiter ID | |
| class TestAsyncioToolsEdgeCases(unittest.TestCase): | |
| def test_frames_without_location_tree(self): | |
| """Frames the unwinder could not fully read - should not crash.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-A", | |
| coroutine_stack=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("<unreadable frame>", "~", None), | |
| FrameInfo("<unknown function>", "app.py", None), | |
| FrameInfo("big", "big.py", None), | |
| ], | |
| task_name=1 | |
| ) | |
| ], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| self.assertEqual( | |
| tools.build_async_tree(input_), | |
| [[ | |
| "└── (T) Task-A", | |
| " └── big big.py", | |
| " └── <unknown function> app.py", | |
| " └── <unreadable frame>", | |
| ]], | |
| ) | |
| def test_frames_without_location_table(self): | |
| """Frame names are not truncated at the first space.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-A", | |
| coroutine_stack=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("<unreadable frame>", "~", None) | |
| ], | |
| task_name=1 | |
| ) | |
| ], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("<unknown function>", "app.py", None) | |
| ], | |
| task_name=2 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| self.assertEqual( | |
| tools.build_task_table(input_), | |
| [[ | |
| 1, | |
| "0x1", | |
| "Task-A", | |
| "<unreadable frame>", | |
| "<unknown function>", | |
| "Unknown", | |
| "0x2", | |
| ]], | |
| ) | |
| def test_task_awaits_self(self): | |
| """A task directly awaits itself - should raise a cycle.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Self-Awaiter", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("loopback", "", LocationInfo(0))], | |
| task_name=1 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| with self.assertRaises(tools.CycleFoundException) as ctx: | |
| tools.build_async_tree(input_) | |
| self.assertIn([1, 1], ctx.exception.cycles) | |
| def test_task_with_missing_awaiter_id(self): | |
| """Awaiter ID not in task list - should not crash, just show 'Unknown'.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-A", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("coro", "", LocationInfo(0))], | |
| task_name=999 | |
| ) | |
| ] | |
| ) | |
| ] | |
| ) | |
| ] | |
| table = tools.build_task_table(input_) | |
| self.assertEqual(len(table), 1) | |
| self.assertEqual(table[0][5], "Unknown") | |
| def test_duplicate_coroutine_frames(self): | |
| """Same coroutine frame repeated under a parent - should deduplicate.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="Task-1", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("frameA", "", LocationInfo(0))], | |
| task_name=2 | |
| ), | |
| CoroInfo( | |
| call_stack=[FrameInfo("frameA", "", LocationInfo(0))], | |
| task_name=3 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name="Task-2", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ), | |
| TaskInfo( | |
| task_id=3, | |
| task_name="Task-3", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| tree = tools.build_async_tree(input_) | |
| # Both children should be under the same coroutine node | |
| flat = "\n".join(tree[0]) | |
| self.assertIn("frameA", flat) | |
| self.assertIn("Task-2", flat) | |
| self.assertIn("Task-1", flat) | |
| flat = "\n".join(tree[1]) | |
| self.assertIn("frameA", flat) | |
| self.assertIn("Task-3", flat) | |
| self.assertIn("Task-1", flat) | |
| def test_task_with_no_name(self): | |
| """Task with no name in id2name - should still render with fallback.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="root", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[FrameInfo("f1", "", LocationInfo(0))], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name=None, | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| # If name is None, fallback to string should not crash | |
| tree = tools.build_async_tree(input_) | |
| self.assertIn("(T) None", "\n".join(tree[0])) | |
| def test_tree_rendering_with_custom_emojis(self): | |
| """Pass custom emojis to the tree renderer.""" | |
| input_ = [ | |
| AwaitedInfo( | |
| thread_id=1, | |
| awaited_by=[ | |
| TaskInfo( | |
| task_id=1, | |
| task_name="MainTask", | |
| coroutine_stack=[], | |
| awaited_by=[ | |
| CoroInfo( | |
| call_stack=[ | |
| FrameInfo("f1", "", LocationInfo(0)), | |
| FrameInfo("f2", "", LocationInfo(0)) | |
| ], | |
| task_name=2 | |
| ) | |
| ] | |
| ), | |
| TaskInfo( | |
| task_id=2, | |
| task_name="SubTask", | |
| coroutine_stack=[], | |
| awaited_by=[] | |
| ) | |
| ] | |
| ) | |
| ] | |
| tree = tools.build_async_tree(input_, task_emoji="🧵", cor_emoji="🔁") | |
| flat = "\n".join(tree[0]) | |
| self.assertIn("🧵 MainTask", flat) | |
| self.assertIn("🔁 f1", flat) | |
| self.assertIn("🔁 f2", flat) | |
| self.assertIn("🧵 SubTask", flat) |