Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pluggy/_callers.py: 17%

1""" 

2Call loop machinery 

3""" 

4 

5from __future__ import annotations 

6 

7from collections.abc import Generator 

8from collections.abc import Mapping 

9from collections.abc import Sequence 

10from typing import cast 

11from typing import NoReturn 

12import warnings 

13 

14from ._hooks import HookImpl 

15from ._result import HookCallError 

16from ._result import Result 

17from ._warnings import PluggyTeardownRaisedWarning 

18 

19 

20# Need to distinguish between old- and new-style hook wrappers. 

21# Wrapping with a tuple is the fastest type-safe way I found to do it. 

22Teardown = Generator[None, object, object] 

23 

24 

25def run_old_style_hookwrapper( 

26 hook_impl: HookImpl, hook_name: str, args: Sequence[object] 

27) -> Teardown: 

28 """ 

29 backward compatibility wrapper to run a old style hookwrapper as a wrapper 

30 """ 

31 

32 teardown: Teardown = cast(Teardown, hook_impl.function(*args)) 

33 try: 

34 next(teardown) 

35 except StopIteration: 

36 _raise_wrapfail(teardown, "did not yield") 

37 try: 

38 res = yield 

39 result = Result(res, None) 

40 except BaseException as exc: 

41 result = Result(None, exc) 

42 try: 

43 teardown.send(result) 

44 except StopIteration: 

45 pass 

46 except BaseException as e: 

47 _warn_teardown_exception(hook_name, hook_impl, e) 

48 raise 

49 else: 

50 _raise_wrapfail(teardown, "has second yield") 

51 finally: 

52 teardown.close() 

53 return result.get_result() 

54 

55 

56def _raise_wrapfail( 

57 wrap_controller: Generator[None, object, object], 

58 msg: str, 

59) -> NoReturn: 

60 co = wrap_controller.gi_code # type: ignore[attr-defined] 

61 raise RuntimeError( 

62 f"wrap_controller at {co.co_name!r} {co.co_filename}:{co.co_firstlineno} {msg}" 

63 ) 

64 

65 

66def _warn_teardown_exception( 

67 hook_name: str, hook_impl: HookImpl, e: BaseException 

68) -> None: 

69 msg = "A plugin raised an exception during an old-style hookwrapper teardown.\n" 

70 msg += f"Plugin: {hook_impl.plugin_name}, Hook: {hook_name}\n" 

71 msg += f"{type(e).__name__}: {e}\n" 

72 msg += "For more information see https://pluggy.readthedocs.io/en/stable/api_reference.html#pluggy.PluggyTeardownRaisedWarning" # noqa: E501 

73 warnings.warn(PluggyTeardownRaisedWarning(msg), stacklevel=6) 

74 

75 

76def _multicall( 

77 hook_name: str, 

78 hook_impls: Sequence[HookImpl], 

79 caller_kwargs: Mapping[str, object], 

80 firstresult: bool, 

81) -> object | list[object]: 

82 """Execute a call into multiple python functions/methods and return the 

83 result(s). 

84 

85 ``caller_kwargs`` comes from HookCaller.__call__(). 

86 """ 

87 __tracebackhide__ = True 

88 results: list[object] = [] 

89 exception = None 

90 try: # run impl and wrapper setup functions in a loop 

91 teardowns: list[Teardown] = [] 

92 try: 

93 for hook_impl in reversed(hook_impls): 

94 try: 

95 args = [caller_kwargs[argname] for argname in hook_impl.argnames] 

96 except KeyError as e: 

97 # coverage bug - this is tested 

98 for argname in hook_impl.argnames: # pragma: no cover 

99 if argname not in caller_kwargs: 

100 raise HookCallError( 

101 f"hook call must provide argument {argname!r}" 

102 ) from e 

103 

104 if hook_impl.hookwrapper: 

105 function_gen = run_old_style_hookwrapper(hook_impl, hook_name, args) 

106 

107 next(function_gen) # first yield 

108 teardowns.append(function_gen) 

109 

110 elif hook_impl.wrapper: 

111 try: 

112 # If this cast is not valid, a type error is raised below, 

113 # which is the desired response. 

114 res = hook_impl.function(*args) 

115 function_gen = cast(Generator[None, object, object], res) 

116 next(function_gen) # first yield 

117 teardowns.append(function_gen) 

118 except StopIteration: 

119 _raise_wrapfail(function_gen, "did not yield") 

120 else: 

121 res = hook_impl.function(*args) 

122 if res is not None: 

123 results.append(res) 

124 if firstresult: # halt further impl calls 

125 break 

126 except BaseException as exc: 

127 exception = exc 

128 finally: 

129 if firstresult: # first result hooks return a single value 

130 result = results[0] if results else None 

131 else: 

132 result = results 

133 

134 # run all wrapper post-yield blocks 

135 for teardown in reversed(teardowns): 

136 try: 

137 if exception is not None: 

138 try: 

139 teardown.throw(exception) 

140 except RuntimeError as re: 

141 # StopIteration from generator causes RuntimeError 

142 # even for coroutine usage - see #544 

143 if ( 

144 isinstance(exception, StopIteration) 

145 and re.__cause__ is exception 

146 ): 

147 teardown.close() 

148 continue 

149 else: 

150 raise 

151 else: 

152 teardown.send(result) 

153 # Following is unreachable for a well behaved hook wrapper. 

154 # Try to force finalizers otherwise postponed till GC action. 

155 # Note: close() may raise if generator handles GeneratorExit. 

156 teardown.close() 

157 except StopIteration as si: 

158 result = si.value 

159 exception = None 

160 continue 

161 except BaseException as e: 

162 exception = e 

163 continue 

164 _raise_wrapfail(teardown, "has second yield") 

165 

166 if exception is not None: 

167 raise exception 

168 else: 

169 return result