1import json
2import re
3from dataclasses import replace
4from functools import cached_property
5from http import HTTPStatus
6from http.cookies import SimpleCookie
7from typing import TYPE_CHECKING, Any
8from urllib.parse import urljoin
9
10from plain.http import Request, WebSocketResponse
11from plain.server.inprocess import HandledRequest, InProcessServer, SentResponse
12
13from .exceptions import RedirectCycleError, require_app
14from .request_builder import (
15 Target,
16 build_encoded_request,
17 encode_request_body,
18 split_target,
19)
20from .websocket import (
21 DEFAULT_TIMEOUT,
22 WebSocketRejected,
23 WebSocketTestConnection,
24 handshake_headers,
25)
26
27if TYPE_CHECKING:
28 from plain.http import Response
29 from plain.http.response import ResponseHeaders
30
31__all__ = [
32 "Client",
33 "ClientResponse",
34]
35
36
37# Structured suffix spec: https://tools.ietf.org/html/rfc6838#section-4.2.8
38_JSON_CONTENT_TYPE_RE = re.compile(r"^application\/(.+\+)?json")
39
40_REDIRECT_STATUS_CODES = (
41 HTTPStatus.MOVED_PERMANENTLY,
42 HTTPStatus.FOUND,
43 HTTPStatus.SEE_OTHER,
44 HTTPStatus.TEMPORARY_REDIRECT,
45 HTTPStatus.PERMANENT_REDIRECT,
46)
47
48
49class ClientResponse:
50 """
51 What the test client got back for one request.
52
53 It has a fixed set of names, listed in `_NAMES`, and reports what was
54 *sent*: the status that went out and the bytes that went out. The
55 `Response` object the app returned is `returned_response`, for the few
56 assertions that are about that object itself.
57 """
58
59 _NAMES = (
60 "status_code",
61 "headers",
62 "cookies",
63 "body",
64 "text",
65 "json_data",
66 "redirect_to",
67 "redirect_chain",
68 "request",
69 "exception",
70 "returned_response",
71 )
72
73 def __init__(self, sent: SentResponse):
74 self._sent = sent
75 self._redirect_chain: list[tuple[str, int]] = []
76
77 @property
78 def status_code(self) -> int:
79 """The status that went out — the view's, unless its streaming body
80 failed before producing anything, which a server answers with 500."""
81 return self._sent.status_code
82
83 @property
84 def headers(self) -> ResponseHeaders:
85 """The response headers."""
86 return self._sent.response.headers
87
88 @property
89 def cookies(self) -> SimpleCookie:
90 """The cookies this response set."""
91 return self._sent.response.cookies
92
93 @property
94 def body(self) -> bytes:
95 """The body the response sent — for a streaming response too, read to
96 the end the way a server sends it (empty for HEAD, 204, 304)."""
97 return self._sent.body
98
99 @property
100 def text(self) -> str:
101 """The body the response sent, decoded as a string."""
102 return self._sent.body.decode(self._sent.response.charset)
103
104 @cached_property
105 def json_data(self) -> Any:
106 """The body the response sent, parsed as JSON (requires a JSON content type)."""
107 content_type = self.headers.get("Content-Type", "")
108 if not _JSON_CONTENT_TYPE_RE.match(content_type):
109 raise ValueError(
110 f'Content-Type header is "{content_type}", not "application/json"'
111 )
112 return json.loads(self.text)
113
114 @property
115 def redirect_to(self) -> str | None:
116 """The redirect target if this is a 3xx response, otherwise None."""
117 if 300 <= self.status_code < 400:
118 return self.headers.get("Location")
119 return None
120
121 @property
122 def redirect_chain(self) -> list[tuple[str, int]]:
123 """The `(url, status_code)` of each redirect `follow_redirects=True`
124 followed to get here. Empty when nothing was followed."""
125 return self._redirect_chain
126
127 @property
128 def request(self) -> Request:
129 """The request that produced this response. The route that handled
130 it is `request.resolver_match`."""
131 return self._sent.request
132
133 @property
134 def exception(self) -> Exception | None:
135 """The exception behind a 5xx response, when
136 `Client(raise_exceptions=False)` kept it from being raised."""
137 return self._sent.response.exception
138
139 @property
140 def returned_response(self) -> Response:
141 """The `Response` object the app returned.
142
143 Everything else here describes what was sent. This is the object
144 itself, for asserting on its type or on an attribute only that type
145 has. Its own `content` and `status_code` can differ from what went
146 out (a HEAD, a 204, a streaming body that failed).
147 """
148 return self._sent.response
149
150 def __getattr__(self, name: str) -> Any:
151 # Only reached for a name that isn't defined above.
152 raise AttributeError(
153 f"The test client's response has no `{name}`. It has: "
154 + ", ".join(type(self)._NAMES)
155 + ". The `Response` the app returned is `response.returned_response`."
156 )
157
158 def __repr__(self) -> str:
159 return (
160 f"<ClientResponse status_code={self._sent.status_code}"
161 f" of {self._sent.response!r}>"
162 )
163
164
165class Client:
166 """
167 A client for making requests against the app without running a server.
168
169 It speaks the same vocabulary as the rest of Plain: `form_data=` arrives
170 as `request.form_data`, `json_data=` as `request.json_data`, `files=` as
171 `request.files`, and `query_params=` as `request.query_params`.
172
173 A path makes a request to `https://testserver`. Pass a full URL when the
174 scheme, host or port matter: `client.get("http://testserver/")`.
175
176 Client objects are stateful — they keep the cookies (and so the session)
177 that responses set, for the lifetime of the Client instance. `cookies` is
178 that jar: logging a client in is writing the session cookie to it, which
179 is what `plain.auth.testing.login_client` does.
180
181 `headers` are sent with every request. `raise_exceptions=False` keeps an
182 exception the app raised as the 5xx response it became, on
183 `response.exception`, where the default is to raise it from the request.
184 """
185
186 def __init__(
187 self,
188 *,
189 raise_exceptions: bool = True,
190 headers: dict[str, str] | None = None,
191 ) -> None:
192 require_app("Client")
193 self.raise_exceptions = raise_exceptions
194 self._headers: dict[str, str] = headers or {}
195 self._cookies: SimpleCookie = SimpleCookie()
196 self._server = InProcessServer()
197
198 @property
199 def cookies(self) -> SimpleCookie:
200 """The cookies sent with every request, updated by every response."""
201 return self._cookies
202
203 def __repr__(self) -> str:
204 # The cookies' names and not their values: a session key is a
205 # credential, and this is printed in failure reports.
206 return f"<Client cookies={sorted(self._cookies)}>"
207
208 def request(
209 self,
210 method: str,
211 path: str,
212 *,
213 query_params: dict[str, Any] | None = None,
214 form_data: dict[str, Any] | None = None,
215 json_data: Any = None,
216 body: bytes | str | None = None,
217 files: dict[str, Any] | None = None,
218 content_type: str | None = None,
219 headers: dict[str, str] | None = None,
220 follow_redirects: bool = False,
221 ) -> ClientResponse:
222 """Make a request with any method."""
223 encoded_body, encoded_content_type = encode_request_body(
224 form_data=form_data,
225 json_data=json_data,
226 body=body,
227 files=files,
228 content_type=content_type,
229 )
230 response = self._send(
231 self._build_request(
232 method,
233 split_target(path, query_params=query_params),
234 body=encoded_body,
235 content_type=encoded_content_type,
236 headers=headers,
237 )
238 )
239 if follow_redirects:
240 response = self._follow_redirects(
241 response,
242 body=encoded_body,
243 content_type=encoded_content_type,
244 headers=headers,
245 )
246 return response
247
248 def _build_request(
249 self,
250 method: str,
251 target: Target,
252 *,
253 body: bytes = b"",
254 content_type: str = "",
255 headers: dict[str, str] | None,
256 ) -> Request:
257 """Build a request as this client sends it: with the client's
258 headers under the request's own, and the cookies in its jar."""
259 all_headers: dict[str, str] = dict(self._headers)
260 if headers:
261 all_headers.update(headers)
262
263 cookie_header = "; ".join(
264 sorted(
265 f"{morsel.key}={morsel.coded_value}"
266 for morsel in self._cookies.values()
267 )
268 )
269 if cookie_header:
270 all_headers["Cookie"] = cookie_header
271
272 return build_encoded_request(
273 method,
274 target,
275 body=body,
276 content_type=content_type,
277 headers=all_headers,
278 )
279
280 def _handle(self, request: Request) -> HandledRequest:
281 """Run a request through the app, and keep the cookies it set.
282
283 Every request this client makes comes through here, a websocket's
284 handshake included.
285 """
286 handled = self._server.handle(request)
287 if handled.response.cookies:
288 self._cookies.update(handled.response.cookies)
289 return handled
290
291 def _sent(self, handled: HandledRequest) -> ClientResponse:
292 """Send a handled request's response and wrap what went out.
293
294 An exception the app raised is raised from here, after the response
295 is sent and closed, unless `raise_exceptions` is off. Only a 5xx
296 has one.
297 """
298 response = ClientResponse(handled.send())
299 if response.exception and self.raise_exceptions:
300 raise response.exception
301 return response
302
303 def _send(self, request: Request) -> ClientResponse:
304 """Run a Request through the app and wrap what came back."""
305 return self._sent(self._handle(request))
306
307 def get(
308 self,
309 path: str,
310 *,
311 query_params: dict[str, Any] | None = None,
312 headers: dict[str, str] | None = None,
313 follow_redirects: bool = False,
314 ) -> ClientResponse:
315 """Make a GET request."""
316 return self.request(
317 "GET",
318 path,
319 query_params=query_params,
320 headers=headers,
321 follow_redirects=follow_redirects,
322 )
323
324 def head(
325 self,
326 path: str,
327 *,
328 query_params: dict[str, Any] | None = None,
329 headers: dict[str, str] | None = None,
330 follow_redirects: bool = False,
331 ) -> ClientResponse:
332 """Make a HEAD request."""
333 return self.request(
334 "HEAD",
335 path,
336 query_params=query_params,
337 headers=headers,
338 follow_redirects=follow_redirects,
339 )
340
341 def options(
342 self,
343 path: str,
344 *,
345 query_params: dict[str, Any] | None = None,
346 headers: dict[str, str] | None = None,
347 follow_redirects: bool = False,
348 ) -> ClientResponse:
349 """Make an OPTIONS request."""
350 return self.request(
351 "OPTIONS",
352 path,
353 query_params=query_params,
354 headers=headers,
355 follow_redirects=follow_redirects,
356 )
357
358 def post(
359 self,
360 path: str,
361 *,
362 form_data: dict[str, Any] | None = None,
363 json_data: Any = None,
364 body: bytes | str | None = None,
365 files: dict[str, Any] | None = None,
366 content_type: str | None = None,
367 query_params: dict[str, Any] | None = None,
368 headers: dict[str, str] | None = None,
369 follow_redirects: bool = False,
370 ) -> ClientResponse:
371 """Make a POST request."""
372 return self.request(
373 "POST",
374 path,
375 form_data=form_data,
376 json_data=json_data,
377 body=body,
378 files=files,
379 content_type=content_type,
380 query_params=query_params,
381 headers=headers,
382 follow_redirects=follow_redirects,
383 )
384
385 def put(
386 self,
387 path: str,
388 *,
389 form_data: dict[str, Any] | None = None,
390 json_data: Any = None,
391 body: bytes | str | None = None,
392 files: dict[str, Any] | None = None,
393 content_type: str | None = None,
394 query_params: dict[str, Any] | None = None,
395 headers: dict[str, str] | None = None,
396 follow_redirects: bool = False,
397 ) -> ClientResponse:
398 """Make a PUT request."""
399 return self.request(
400 "PUT",
401 path,
402 form_data=form_data,
403 json_data=json_data,
404 body=body,
405 files=files,
406 content_type=content_type,
407 query_params=query_params,
408 headers=headers,
409 follow_redirects=follow_redirects,
410 )
411
412 def patch(
413 self,
414 path: str,
415 *,
416 form_data: dict[str, Any] | None = None,
417 json_data: Any = None,
418 body: bytes | str | None = None,
419 files: dict[str, Any] | None = None,
420 content_type: str | None = None,
421 query_params: dict[str, Any] | None = None,
422 headers: dict[str, str] | None = None,
423 follow_redirects: bool = False,
424 ) -> ClientResponse:
425 """Make a PATCH request."""
426 return self.request(
427 "PATCH",
428 path,
429 form_data=form_data,
430 json_data=json_data,
431 body=body,
432 files=files,
433 content_type=content_type,
434 query_params=query_params,
435 headers=headers,
436 follow_redirects=follow_redirects,
437 )
438
439 def delete(
440 self,
441 path: str,
442 *,
443 form_data: dict[str, Any] | None = None,
444 json_data: Any = None,
445 body: bytes | str | None = None,
446 files: dict[str, Any] | None = None,
447 content_type: str | None = None,
448 query_params: dict[str, Any] | None = None,
449 headers: dict[str, str] | None = None,
450 follow_redirects: bool = False,
451 ) -> ClientResponse:
452 """Make a DELETE request."""
453 return self.request(
454 "DELETE",
455 path,
456 form_data=form_data,
457 json_data=json_data,
458 body=body,
459 files=files,
460 content_type=content_type,
461 query_params=query_params,
462 headers=headers,
463 follow_redirects=follow_redirects,
464 )
465
466 def websocket(
467 self,
468 path: str,
469 *,
470 subprotocols: tuple[str, ...] = (),
471 query_params: dict[str, Any] | None = None,
472 headers: dict[str, str] | None = None,
473 timeout: float = DEFAULT_TIMEOUT,
474 ) -> WebSocketTestConnection:
475 """Open a websocket to `path` and drive the view's `websocket()` in-process.
476
477 with client.websocket("/live/", subprotocols=("binary",)) as ws:
478 ws.send(b"...")
479 assert ws.receive() == b"..."
480
481 The handshake runs through the normal pipeline with this client's
482 cookies, so auth applies. A response other than the 101 raises
483 `WebSocketRejected` carrying it, and a handshake the app raised from
484 raises that exception, as any other request does. Every call on the
485 connection has a timeout (default 5 s) and raises `TimeoutError`
486 when it elapses.
487 """
488 handshake = handshake_headers(subprotocols=subprotocols)
489 if headers:
490 handshake.update(headers)
491
492 handled = self._handle(
493 self._build_request(
494 "GET",
495 split_target(path, query_params=query_params),
496 headers=handshake,
497 )
498 )
499 if isinstance(handled.response, WebSocketResponse):
500 return WebSocketTestConnection(handled, timeout=timeout)
501 raise WebSocketRejected(self._sent(handled))
502
503 def _follow_redirects(
504 self,
505 response: ClientResponse,
506 *,
507 body: bytes,
508 content_type: str,
509 headers: dict[str, str] | None,
510 ) -> ClientResponse:
511 """
512 Follow redirect responses until a non-redirect response is reached.
513 """
514 redirect_chain: list[tuple[str, int]] = []
515 while response.status_code in _REDIRECT_STATUS_CODES:
516 location = response.redirect_to
517 if location is None:
518 break # a 3xx without a Location header — nowhere to go
519 redirect_chain.append((location, response.status_code))
520
521 previous = response.request
522 target = _redirect_target(location, previous=previous)
523
524 method = previous.method
525 if response.status_code in (
526 HTTPStatus.TEMPORARY_REDIRECT,
527 HTTPStatus.PERMANENT_REDIRECT,
528 ) and method not in ("GET", "HEAD"):
529 # 307/308 preserve the request method and body.
530 request = self._build_request(
531 method,
532 target,
533 body=body,
534 content_type=content_type,
535 headers=headers,
536 )
537 else:
538 # Everything else redirects as a GET without a body.
539 request = self._build_request(
540 "GET" if method not in ("GET", "HEAD") else method,
541 target,
542 headers=headers,
543 )
544 body = b""
545 content_type = ""
546
547 response = self._send(request)
548 response._redirect_chain = redirect_chain
549
550 if redirect_chain[-1] in redirect_chain[:-1]:
551 # Check that we're not redirecting to somewhere we've already
552 # been to, to prevent loops.
553 raise RedirectCycleError(
554 "Redirect loop detected.", last_response=response
555 )
556 if len(redirect_chain) > 20:
557 # Such a lengthy chain likely also means a loop, but one with
558 # a growing path, changing view, or changing query argument;
559 # 20 is the value of "network.http.redirection-limit" from Firefox.
560 raise RedirectCycleError("Too many redirects.", last_response=response)
561
562 return response
563
564
565def _redirect_target(location: str, *, previous: Request) -> Target:
566 """Where a `Location` header leads, from the request that was redirected.
567
568 As a browser resolves it: whatever the location leaves out is the
569 previous request's. One that names a scheme or a host without a port
570 goes to that scheme's port, not the previous request's.
571 """
572 previous_url = (
573 f"{previous.scheme}://{previous.server_name}:{previous.server_port}"
574 f"{previous.path}"
575 )
576 target = split_target(urljoin(previous_url, location))
577 if not target.path:
578 # RFC 3986 Section 6.2.3: Empty path should be normalized to "/".
579 return replace(target, path="/")
580 return target