Coverage for src/zapy/requests/requester.py: 94%

1import asyncio 

2import inspect 

3from dataclasses import dataclass, field 

4from typing import Any, Callable, List 

5 

6import httpx 

7 

8from zapy.base import Metadata 

9from zapy.store import Store, use_store 

10from zapy.templating.traceback import annotate_traceback 

11from zapy.test import TestResult, run_tests 

12from zapy.utils import functools 

13 

14from .context import ZapyRequestContext 

15from .converter import RequestConverter 

16from .exceptions import RenderLocationError, error_location 

17from .hooks import RequestHook, use_global_hook 

18from .models import HttpxArguments, HttpxResponse, ZapyRequest 

19 

20 

21@dataclass 

22class RequesterResponse: 

23 response: httpx.Response 

24 test_result: List[TestResult] = field(default_factory=list) 

25 

26 

27_http_request_signature = inspect.signature(httpx.Client.build_request).parameters 

28 

29 

30class Requester: 

31 

32 def __init__(self, zapy_request: ZapyRequest, converter: RequestConverter, client: httpx.AsyncClient): 

33 self.zapy_request = zapy_request 

34 self.converter = converter 

35 self.client = client 

36 self.__hook_context = { 

37 Metadata: self.zapy_request.metadata, 

38 ZapyRequest: self.zapy_request, 

39 } 

40 

41 async def make_request(self) -> RequesterResponse: 

42 httpx_args = self.converter.build_httpx_args() 

43 

44 # Hook: pre_request 

45 await self._invoke_hooks_pre_request(httpx_args) 

46 

47 # send request 

48 request_parameters, rest_parameters = self._split_parameters(httpx_args) 

49 request = self.client.build_request(**request_parameters) 

50 response = await self.client.send(request, **rest_parameters) 

51 

52 # Hook: post_request 

53 try: 

54 await self._invoke_hooks_post_request(response) 

55 except RenderLocationError as ex: 

56 ex.context["response"] = response 

57 raise 

58 

59 response_wrapper = RequesterResponse(response) 

60 

61 # Hook: test 

62 if self.request_hooks.test: 

63 response_wrapper.test_result = self._run_test( 

64 httpx_args=httpx_args, 

65 request=request, 

66 response=response, 

67 ) 

68 

69 return response_wrapper 

70 

71 @error_location("pre_request") 

72 async def _invoke_hooks_pre_request(self, httpx_args: HttpxArguments) -> None: 

73 try: 

74 await self.__call_hook(use_global_hook().pre_request, httpx_args) 

75 await self.__call_hook(self.request_hooks.pre_request, httpx_args) 

76 except BaseException as e: 

77 annotate_traceback(e, self.converter.script, location="hook") 

78 raise e 

79 

80 @error_location("post_request") 

81 async def _invoke_hooks_post_request(self, response: httpx.Response) -> None: 

82 try: 

83 await self.__call_hook(use_global_hook().post_request, response) 

84 await self.__call_hook(self.request_hooks.post_request, response) 

85 except Exception as e: 

86 annotate_traceback(e, self.converter.script, location="hook") 

87 raise e 

88 

89 def _run_test( 

90 self, httpx_args: HttpxArguments, request: httpx.Request, response: HttpxResponse 

91 ) -> list[TestResult]: 

92 if self.request_hooks.test is None: 92 ↛ 93line 92 didn't jump to line 93, because the condition on line 92 was never true

93 return [] 

94 

95 class RequestMeta(type): 

96 def __new__(cls, name: str, bases: tuple, attrs: dict) -> type: 

97 attrs["httpx_args"] = httpx_args 

98 attrs["request"] = request 

99 attrs["response"] = response 

100 return super().__new__(cls, name, bases, attrs) 

101 

102 class DecoratedClass(self.request_hooks.test, metaclass=RequestMeta): # type: ignore[name-defined] 

103 pass 

104 

105 test_result = run_tests(DecoratedClass).as_list() 

106 return test_result 

107 

108 def _split_parameters(self, httpx_args: HttpxArguments) -> tuple[dict, dict]: 

109 request_parameters, rest_parameters = {}, {} 

110 for k, v in httpx_args.items(): 

111 if k in _http_request_signature: 111 ↛ 114line 111 didn't jump to line 114, because the condition on line 111 was never false

112 request_parameters[k] = v 

113 else: 

114 rest_parameters[k] = v 

115 return request_parameters, rest_parameters 

116 

117 async def __call_hook(self, hook: Callable, *args: Any) -> Any: 

118 result = functools.call_with_signature(hook, *args, kwargs=self.__hook_context) 

119 if asyncio.iscoroutine(result): 119 ↛ 121line 119 didn't jump to line 121, because the condition on line 119 was never false

120 return await result 

121 return result 

122 

123 @property 

124 def request_hooks(self) -> RequestHook: 

125 return self.converter.request_hooks 

126 

127 

128async def send_request( 

129 zapy_request: ZapyRequest, 

130 *, 

131 store: Store | None = None, 

132 logger: Callable = print, 

133 client: httpx.AsyncClient | None = None, 

134) -> RequesterResponse: 

135 if store is None: 

136 store = use_store() 

137 ctx = ZapyRequestContext( 

138 store=store, 

139 logger=logger, 

140 ) 

141 _client = client or httpx.AsyncClient(follow_redirects=True, timeout=None) 

142 try: 

143 request = build_request(zapy_request, ctx, client=_client) 

144 return await request.make_request() 

145 finally: 

146 if client is None: 146 ↛ exitline 146 didn't return from function 'send_request', because the return on line 144 wasn't executed

147 await _client.aclose() 

148 

149 

150def build_request(zapy_request: ZapyRequest, ctx: ZapyRequestContext, client: httpx.AsyncClient) -> Requester: 

151 converter = RequestConverter(zapy_request, ctx) 

152 requester = Requester(zapy_request, converter, client) 

153 

154 return requester