Coverage for nexios\testing\client.py: 100%

1# type: ignore 

2 

3import httpx 

4import typing 

5import uuid 

6from typing import Any, Dict, AsyncIterable, Iterable, Union, Optional, List 

7from .transport import ( 

8 NexiosAsyncTransport, 

9 WebSocketConnection, 

10 WebSocketState, 

11 WebSocketDisconnect, 

12) 

13 

14_RequestData = typing.Mapping[str, typing.Union[str, typing.Iterable[str], bytes]] 

15from nexios.application import NexiosApp 

16 

17 

18class Client(httpx.AsyncClient): 

19 def __init__( 

20 self, 

21 app: NexiosApp, 

22 root_path: str = "", 

23 client: tuple[str, int] = ("testclient", 5000), 

24 base_url: str = "http://testserver", 

25 raise_server_exceptions: bool = True, 

26 cookies: Union[httpx._types.CookieTypes, None] = None, # type: ignore 

27 headers: Union[Dict[str, str], None] = None, 

28 follow_redirects: bool = True, 

29 max_retries: int = 3, 

30 timeout: Union[httpx._types.TimeoutTypes, float] = 5.0, # type: ignore 

31 log_requests: bool = False, 

32 app_state: Dict[str, Any] = {}, 

33 **kwargs: Any, 

34 ) -> None: 

35 if headers is None: 

36 headers = {} 

37 headers.setdefault("user-agent", "testclient") 

38 transport = NexiosAsyncTransport( 

39 app=app, 

40 app_state=app_state, 

41 raise_app_exceptions=raise_server_exceptions, 

42 root_path=root_path, 

43 client=client, 

44 ) 

45 super().__init__( 

46 base_url=base_url, 

47 headers=headers, 

48 follow_redirects=follow_redirects, 

49 cookies=cookies, 

50 timeout=timeout, 

51 transport=transport, 

52 **kwargs, 

53 ) 

54 

55 self.max_retries = max_retries 

56 self.log_requests = log_requests 

57 

58 async def handle_request( 

59 self, 

60 method: str, 

61 url: httpx._types.URLTypes, # type: ignore 

62 *, 

63 content: Union[str, bytes, Iterable[bytes], AsyncIterable[bytes], None] = None, # type: ignore 

64 data: Union[_RequestData, None] = None, 

65 files: Union[httpx._types.RequestFiles, None] = None, # type: ignore 

66 json: typing.Any = None, 

67 params: Union[httpx._types.QueryParamTypes, None] = None, # type: ignore 

68 headers: Union[httpx._types.HeaderTypes, None] = None, # type: ignore 

69 cookies: Union[httpx._types.CookieTypes, None] = None, # type: ignore 

70 auth: Union[httpx._types.AuthTypes, httpx._client.UseClientDefault] = httpx._client.USE_CLIENT_DEFAULT, # type: ignore 

71 follow_redirects: Union[bool, httpx._client.UseClientDefault] = httpx._client.USE_CLIENT_DEFAULT, # type: ignore 

72 timeout: Union[httpx._types.TimeoutTypes, httpx._client.UseClientDefault] = httpx._client.USE_CLIENT_DEFAULT, # type: ignore 

73 extensions: Union[Dict[str, typing.Any], None] = None, 

74 ) -> httpx.Response: 

75 retries = 0 

76 last_exception = None 

77 

78 response = await super().request( 

79 method, 

80 url, 

81 content=content, 

82 data=data, 

83 files=files, 

84 json=json, 

85 params=params, 

86 headers=headers, 

87 cookies=cookies, 

88 auth=auth, 

89 follow_redirects=follow_redirects, 

90 timeout=timeout, 

91 extensions=extensions, 

92 ) 

93 return response 

94 

95 async def get( 

96 self, 

97 url: httpx._types.URLTypes, 

98 *, 

99 params: Union[httpx._types.QueryParamTypes, None] = None, # type: ignore 

100 headers: Union[httpx._types.HeaderTypes, None] = None, # type: ignore 

101 cookies: Union[httpx._types.CookieTypes, None] = None, # type: ignore 

102 auth: Union[httpx._types.AuthTypes, httpx._client.UseClientDefault] = httpx._client.USE_CLIENT_DEFAULT, # type: ignore 

103 follow_redirects: Union[bool, httpx._client.UseClientDefault] = httpx._client.USE_CLIENT_DEFAULT, # type: ignore 

104 timeout: Union[httpx._types.TimeoutTypes, httpx._client.UseClientDefault] = httpx._client.USE_CLIENT_DEFAULT, # type: ignore 

105 extensions: Union[Dict[str, typing.Any], None] = None, 

106 ) -> httpx.Response: 

107 return await self.handle_request( 

108 "GET", 

109 url, 

110 params=params, 

111 headers=headers, 

112 cookies=cookies, 

113 auth=auth, 

114 follow_redirects=follow_redirects, 

115 timeout=timeout, 

116 extensions=extensions, 

117 ) 

118 

119 async def post( 

120 self, 

121 url: httpx._types.URLTypes, 

122 *, 

123 content: Union[httpx._types.RequestContent, None] = None, 

124 data: Union[_RequestData, None] = None, 

125 files: Union[httpx._types.RequestFiles, None] = None, 

126 json: typing.Any = None, 

127 params: Union[httpx._types.QueryParamTypes, None] = None, 

128 headers: Union[httpx._types.HeaderTypes, None] = None, 

129 cookies: Union[httpx._types.CookieTypes, None] = None, 

130 auth: Union[ 

131 httpx._types.AuthTypes, httpx._client.UseClientDefault 

132 ] = httpx._client.USE_CLIENT_DEFAULT, 

133 follow_redirects: Union[ 

134 bool, httpx._client.UseClientDefault 

135 ] = httpx._client.USE_CLIENT_DEFAULT, 

136 timeout: Union[ 

137 httpx._types.TimeoutTypes, httpx._client.UseClientDefault 

138 ] = httpx._client.USE_CLIENT_DEFAULT, 

139 extensions: Union[Dict[str, typing.Any], None] = None, 

140 ) -> httpx.Response: 

141 return await self.handle_request( 

142 "POST", 

143 url, 

144 content=content, 

145 data=data, 

146 files=files, 

147 json=json, 

148 params=params, 

149 headers=headers, 

150 cookies=cookies, 

151 auth=auth, 

152 follow_redirects=follow_redirects, 

153 timeout=timeout, 

154 extensions=extensions, 

155 ) 

156 

157 async def __aenter__(self) -> "Client": 

158 await super().__aenter__() 

159 return self 

160 

161 async def __aexit__(self, *args: typing.Any) -> None: 

162 await super().__aexit__(*args)