-
-
Notifications
You must be signed in to change notification settings - Fork 422
/
client.py
214 lines (178 loc) · 6.68 KB
/
client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
from json import dumps as json_dumps
from json import loads as json_loads
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from unittest.mock import Mock
from urllib.parse import urljoin
from django.http import QueryDict, StreamingHttpResponse
from django.http.request import HttpHeaders, HttpRequest
from ninja import NinjaAPI, Router
from ninja.responses import NinjaJSONEncoder
from ninja.responses import Response as HttpResponse
def build_absolute_uri(location: Optional[str] = None) -> str:
base = "http://testlocation/"
if location:
base = urljoin(base, location)
return base
# TODO: this should be changed
# maybe add here urlconf object and add urls from here
class NinjaClientBase:
__test__ = False # <- skip pytest
def __init__(
self,
router_or_app: Union[NinjaAPI, Router],
headers: Optional[Dict[str, str]] = None,
COOKIES: Optional[Dict[str, str]] = None,
) -> None:
self.headers = headers or {}
self.cookies = COOKIES or {}
self.router_or_app = router_or_app
def get(
self, path: str, data: Optional[Dict] = None, **request_params: Any
) -> "NinjaResponse":
return self.request("GET", path, data, **request_params)
def post(
self,
path: str,
data: Optional[Dict] = None,
json: Any = None,
**request_params: Any,
) -> "NinjaResponse":
return self.request("POST", path, data, json, **request_params)
def patch(
self,
path: str,
data: Optional[Dict] = None,
json: Any = None,
**request_params: Any,
) -> "NinjaResponse":
return self.request("PATCH", path, data, json, **request_params)
def put(
self,
path: str,
data: Optional[Dict] = None,
json: Any = None,
**request_params: Any,
) -> "NinjaResponse":
return self.request("PUT", path, data, json, **request_params)
def delete(
self,
path: str,
data: Optional[Dict] = None,
json: Any = None,
**request_params: Any,
) -> "NinjaResponse":
return self.request("DELETE", path, data, json, **request_params)
def request(
self,
method: str,
path: str,
data: Optional[Dict] = None,
json: Any = None,
**request_params: Any,
) -> "NinjaResponse":
if json is not None:
request_params["body"] = json_dumps(json, cls=NinjaJSONEncoder)
if data is None:
data = {}
if self.headers or request_params.get("headers"):
request_params["headers"] = {
**self.headers,
**request_params.get("headers", {}),
}
if self.cookies or request_params.get("COOKIES"):
request_params["COOKIES"] = {
**self.cookies,
**request_params.get("COOKIES", {}),
}
func, request, kwargs = self._resolve(method, path, data, request_params)
return self._call(func, request, kwargs) # type: ignore
@property
def urls(self) -> List:
if not hasattr(self, "_urls_cache"):
self._urls_cache: List
if isinstance(self.router_or_app, NinjaAPI):
self._urls_cache = self.router_or_app.urls[0]
else:
api = NinjaAPI()
self.router_or_app.set_api_instance(api)
self._urls_cache = list(self.router_or_app.urls_paths(""))
return self._urls_cache
def _resolve(
self, method: str, path: str, data: Dict, request_params: Any
) -> Tuple[Callable, Mock, Dict]:
url_path = path.split("?")[0].lstrip("/")
for url in self.urls:
match = url.resolve(url_path)
if match:
request = self._build_request(method, path, data, request_params)
return match.func, request, match.kwargs
raise Exception(f'Cannot resolve "{path}"')
def _build_request(
self, method: str, path: str, data: Dict, request_params: Any
) -> Mock:
request = Mock(spec=HttpRequest)
request.method = method
request.path = path
request.body = ""
request.COOKIES = {}
request._dont_enforce_csrf_checks = True
request.is_secure.return_value = False
request.build_absolute_uri = build_absolute_uri
request.auth = None
request.user = Mock()
if "user" not in request_params:
request.user.is_authenticated = False
request.META = request_params.pop("META", {"REMOTE_ADDR": "127.0.0.1"})
request.FILES = request_params.pop("FILES", {})
request.META.update(
{
f"HTTP_{k.replace('-', '_')}": v
for k, v in request_params.pop("headers", {}).items()
}
)
request.headers = HttpHeaders(request.META)
if isinstance(data, QueryDict):
request.POST = data
else:
request.POST = QueryDict(mutable=True)
if isinstance(data, (str, bytes)):
request_params["body"] = data
elif data:
for k, v in data.items():
request.POST[k] = v
if "?" in path:
request.GET = QueryDict(path.split("?")[1])
else:
request.GET = QueryDict()
for k, v in request_params.items():
setattr(request, k, v)
return request
class TestClient(NinjaClientBase):
def _call(self, func: Callable, request: Mock, kwargs: Dict) -> "NinjaResponse":
return NinjaResponse(func(request, **kwargs))
class TestAsyncClient(NinjaClientBase):
async def _call(
self, func: Callable, request: Mock, kwargs: Dict
) -> "NinjaResponse":
return NinjaResponse(await func(request, **kwargs))
class NinjaResponse:
def __init__(self, http_response: Union[HttpResponse, StreamingHttpResponse]):
self._response = http_response
self.status_code = http_response.status_code
self.streaming = http_response.streaming
if self.streaming:
self.content = b"".join(http_response.streaming_content) # type: ignore
else:
self.content = http_response.content # type: ignore[union-attr]
self._data = None
def json(self) -> Any:
return json_loads(self.content)
@property
def data(self) -> Any:
if self._data is None: # Recomputes if json() is None but cheap then
self._data = self.json()
return self._data
def __getitem__(self, key: str) -> Any:
return self._response[key]
def __getattr__(self, attr: str) -> Any:
return getattr(self._response, attr)