ยปCore Development>Code coverage>Lib/asyncio/base_tasks.py

Python code coverage for Lib/asyncio/base_tasks.py

#countcontent
1n/aimport linecache
2n/aimport traceback
3n/a
4n/afrom . import base_futures
5n/afrom . import coroutines
6n/a
7n/a
8n/adef _task_repr_info(task):
9n/a info = base_futures._future_repr_info(task)
10n/a
11n/a if task._must_cancel:
12n/a # replace status
13n/a info[0] = 'cancelling'
14n/a
15n/a coro = coroutines._format_coroutine(task._coro)
16n/a info.insert(1, 'coro=<%s>' % coro)
17n/a
18n/a if task._fut_waiter is not None:
19n/a info.insert(2, 'wait_for=%r' % task._fut_waiter)
20n/a return info
21n/a
22n/a
23n/adef _task_get_stack(task, limit):
24n/a frames = []
25n/a try:
26n/a # 'async def' coroutines
27n/a f = task._coro.cr_frame
28n/a except AttributeError:
29n/a f = task._coro.gi_frame
30n/a if f is not None:
31n/a while f is not None:
32n/a if limit is not None:
33n/a if limit <= 0:
34n/a break
35n/a limit -= 1
36n/a frames.append(f)
37n/a f = f.f_back
38n/a frames.reverse()
39n/a elif task._exception is not None:
40n/a tb = task._exception.__traceback__
41n/a while tb is not None:
42n/a if limit is not None:
43n/a if limit <= 0:
44n/a break
45n/a limit -= 1
46n/a frames.append(tb.tb_frame)
47n/a tb = tb.tb_next
48n/a return frames
49n/a
50n/a
51n/adef _task_print_stack(task, limit, file):
52n/a extracted_list = []
53n/a checked = set()
54n/a for f in task.get_stack(limit=limit):
55n/a lineno = f.f_lineno
56n/a co = f.f_code
57n/a filename = co.co_filename
58n/a name = co.co_name
59n/a if filename not in checked:
60n/a checked.add(filename)
61n/a linecache.checkcache(filename)
62n/a line = linecache.getline(filename, lineno, f.f_globals)
63n/a extracted_list.append((filename, lineno, name, line))
64n/a exc = task._exception
65n/a if not extracted_list:
66n/a print('No stack for %r' % task, file=file)
67n/a elif exc is not None:
68n/a print('Traceback for %r (most recent call last):' % task,
69n/a file=file)
70n/a else:
71n/a print('Stack for %r (most recent call last):' % task,
72n/a file=file)
73n/a traceback.print_list(extracted_list, file=file)
74n/a if exc is not None:
75n/a for line in traceback.format_exception_only(exc.__class__, exc):
76n/a print(line, file=file, end='')