Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/anyio/_core/_testing.py: 44%

1from typing import Any, Awaitable, Generator, Optional, Union 

2 

3from ._compat import DeprecatedAwaitableList, _warn_deprecation 

4from ._eventloop import get_asynclib 

5 

6 

7class TaskInfo: 

8 """ 

9 Represents an asynchronous task. 

10 

11 :ivar int id: the unique identifier of the task 

12 :ivar parent_id: the identifier of the parent task, if any 

13 :vartype parent_id: Optional[int] 

14 :ivar str name: the description of the task (if any) 

15 :ivar ~collections.abc.Coroutine coro: the coroutine object of the task 

16 """ 

17 

18 __slots__ = "_name", "id", "parent_id", "name", "coro" 

19 

20 def __init__( 

21 self, 

22 id: int, 

23 parent_id: Optional[int], 

24 name: Optional[str], 

25 coro: Union[Generator, Awaitable[Any]], 

26 ): 

27 func = get_current_task 

28 self._name = f"{func.__module__}.{func.__qualname__}" 

29 self.id: int = id 

30 self.parent_id: Optional[int] = parent_id 

31 self.name: Optional[str] = name 

32 self.coro: Union[Generator, Awaitable[Any]] = coro 

33 

34 def __eq__(self, other: object) -> bool: 

35 if isinstance(other, TaskInfo): 

36 return self.id == other.id 

37 

38 return NotImplemented 

39 

40 def __hash__(self) -> int: 

41 return hash(self.id) 

42 

43 def __repr__(self) -> str: 

44 return f"{self.__class__.__name__}(id={self.id!r}, name={self.name!r})" 

45 

46 def __await__(self) -> Generator[None, None, "TaskInfo"]: 

47 _warn_deprecation(self) 

48 if False: 

49 yield 

50 

51 return self 

52 

53 def _unwrap(self) -> "TaskInfo": 

54 return self 

55 

56 

57def get_current_task() -> TaskInfo: 

58 """ 

59 Return the current task. 

60 

61 :return: a representation of the current task 

62 

63 """ 

64 return get_asynclib().get_current_task() 

65 

66 

67def get_running_tasks() -> DeprecatedAwaitableList[TaskInfo]: 

68 """ 

69 Return a list of running tasks in the current event loop. 

70 

71 :return: a list of task info objects 

72 

73 """ 

74 tasks = get_asynclib().get_running_tasks() 

75 return DeprecatedAwaitableList(tasks, func=get_running_tasks) 

76 

77 

78async def wait_all_tasks_blocked() -> None: 

79 """Wait until all other tasks are waiting for something.""" 

80 await get_asynclib().wait_all_tasks_blocked()