(st: FutureCallGraph, buf: list[str], level: int)
| 195 | """ |
| 196 | |
| 197 | def render_level(st: FutureCallGraph, buf: list[str], level: int) -> None: |
| 198 | def add_line(line: str) -> None: |
| 199 | buf.append(level * ' ' + line) |
| 200 | |
| 201 | if isinstance(st.future, tasks.Task): |
| 202 | add_line( |
| 203 | f'* Task(name={st.future.get_name()!r}, id={id(st.future):#x})' |
| 204 | ) |
| 205 | else: |
| 206 | add_line( |
| 207 | f'* Future(id={id(st.future):#x})' |
| 208 | ) |
| 209 | |
| 210 | if st.call_stack: |
| 211 | add_line( |
| 212 | f' + Call stack:' |
| 213 | ) |
| 214 | for ste in st.call_stack: |
| 215 | f = ste.frame |
| 216 | |
| 217 | if f.f_generator is None: |
| 218 | f = ste.frame |
| 219 | add_line( |
| 220 | f' | File {f.f_code.co_filename!r},' |
| 221 | f' line {f.f_lineno}, in' |
| 222 | f' {f.f_code.co_qualname}()' |
| 223 | ) |
| 224 | else: |
| 225 | c = f.f_generator |
| 226 | |
| 227 | try: |
| 228 | f = c.cr_frame |
| 229 | code = c.cr_code |
| 230 | tag = 'async' |
| 231 | except AttributeError: |
| 232 | try: |
| 233 | f = c.ag_frame |
| 234 | code = c.ag_code |
| 235 | tag = 'async generator' |
| 236 | except AttributeError: |
| 237 | f = c.gi_frame |
| 238 | code = c.gi_code |
| 239 | tag = 'generator' |
| 240 | |
| 241 | add_line( |
| 242 | f' | File {f.f_code.co_filename!r},' |
| 243 | f' line {f.f_lineno}, in' |
| 244 | f' {tag} {code.co_qualname}()' |
| 245 | ) |
| 246 | |
| 247 | if st.awaited_by: |
| 248 | add_line( |
| 249 | f' + Awaited by:' |
| 250 | ) |
| 251 | for fut in st.awaited_by: |
| 252 | render_level(fut, buf, level + 1) |
| 253 | |
| 254 | graph = capture_call_graph(future, depth=depth + 1, limit=limit) |
no test coverage detected
searching dependent graphs…