Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/mock/backports.py: 14%

1import sys 

2 

3 

4if sys.version_info[:2] > (3, 9): 

5 from inspect import iscoroutinefunction 

6elif sys.version_info[:2] >= (3, 8): 

7 from asyncio import iscoroutinefunction 

8else: 

9 

10 import functools 

11 from asyncio.coroutines import _is_coroutine 

12 from inspect import ismethod, isfunction, CO_COROUTINE 

13 

14 def _unwrap_partial(func): 

15 while isinstance(func, functools.partial): 

16 func = func.func 

17 return func 

18 

19 def _has_code_flag(f, flag): 

20 """Return true if ``f`` is a function (or a method or functools.partial 

21 wrapper wrapping a function) whose code object has the given ``flag`` 

22 set in its flags.""" 

23 while ismethod(f): 

24 f = f.__func__ 

25 f = _unwrap_partial(f) 

26 if not isfunction(f): 

27 return False 

28 return bool(f.__code__.co_flags & flag) 

29 

30 def iscoroutinefunction(obj): 

31 """Return true if the object is a coroutine function. 

32 

33 Coroutine functions are defined with "async def" syntax. 

34 """ 

35 return ( 

36 _has_code_flag(obj, CO_COROUTINE) or 

37 getattr(obj, '_is_coroutine', None) is _is_coroutine 

38 ) 

39 

40 

41try: 

42 from unittest import IsolatedAsyncioTestCase 

43except ImportError: 

44 import asyncio 

45 from unittest import TestCase 

46 

47 

48 class IsolatedAsyncioTestCase(TestCase): 

49 

50 def __init__(self, methodName='runTest'): 

51 super().__init__(methodName) 

52 self._asyncioTestLoop = None 

53 self._asyncioCallsQueue = None 

54 

55 async def _asyncioLoopRunner(self, fut): 

56 self._asyncioCallsQueue = queue = asyncio.Queue() 

57 fut.set_result(None) 

58 while True: 

59 query = await queue.get() 

60 queue.task_done() 

61 assert query is None 

62 

63 def _setupAsyncioLoop(self): 

64 assert self._asyncioTestLoop is None 

65 loop = asyncio.new_event_loop() 

66 asyncio.set_event_loop(loop) 

67 loop.set_debug(True) 

68 self._asyncioTestLoop = loop 

69 fut = loop.create_future() 

70 self._asyncioCallsTask = loop.create_task(self._asyncioLoopRunner(fut)) 

71 loop.run_until_complete(fut) 

72 

73 def _tearDownAsyncioLoop(self): 

74 assert self._asyncioTestLoop is not None 

75 loop = self._asyncioTestLoop 

76 self._asyncioTestLoop = None 

77 self._asyncioCallsQueue.put_nowait(None) 

78 loop.run_until_complete(self._asyncioCallsQueue.join()) 

79 

80 try: 

81 # shutdown asyncgens 

82 loop.run_until_complete(loop.shutdown_asyncgens()) 

83 finally: 

84 asyncio.set_event_loop(None) 

85 loop.close() 

86 

87 def run(self, result=None): 

88 self._setupAsyncioLoop() 

89 try: 

90 return super().run(result) 

91 finally: 

92 self._tearDownAsyncioLoop() 

93 

94 

95try: 

96 from asyncio import _set_event_loop_policy as set_event_loop_policy 

97except ImportError: 

98 from asyncio import set_event_loop_policy