1"""Building a `Request` without sending it.
2
3`build_request()` is the public function. The two it is made of are what
4`Client` uses as well, so a request the client sends is encoded and built
5exactly as one a test builds by hand.
6"""
7
8import json
9from dataclasses import dataclass
10from typing import Any
11from urllib.parse import urlsplit
12
13from plain.http import Request
14from plain.json import PlainJSONEncoder
15from plain.utils.http import parse_header_parameters, urlencode
16
17from .encoding import encode_multipart
18
19__all__ = ["build_request"]
20
21_BOUNDARY = "BoUnDaRyStRiNg"
22_MULTIPART_CONTENT = f"multipart/form-data; boundary={_BOUNDARY}"
23
24# Where a request goes when its path doesn't say: https://testserver.
25DEFAULT_SCHEME = "https"
26DEFAULT_HOST = "testserver"
27DEFAULT_PORTS = {"https": "443", "http": "80"}
28
29
30def build_request(
31 method: str,
32 path: str,
33 *,
34 query_params: dict[str, Any] | None = None,
35 form_data: dict[str, Any] | None = None,
36 json_data: Any = None,
37 body: bytes | str | None = None,
38 files: dict[str, Any] | None = None,
39 content_type: str | None = None,
40 headers: dict[str, str] | None = None,
41) -> Request:
42 """
43 Build a `Request` without sending it, to hand to a view or a middleware
44 directly.
45
46 request = build_request("POST", "/submit", form_data={"name": "Ada"})
47
48 It takes the keywords `Client`'s methods take, and encodes the body the
49 same way.
50
51 `path` is a path, which makes a request to `https://testserver`. Pass a
52 full URL when the scheme, host or port matter:
53
54 request = build_request("GET", "http://testserver/") # not HTTPS
55 """
56 encoded_body, encoded_content_type = encode_request_body(
57 form_data=form_data,
58 json_data=json_data,
59 body=body,
60 files=files,
61 content_type=content_type,
62 )
63 return build_encoded_request(
64 method,
65 split_target(path, query_params=query_params),
66 body=encoded_body,
67 content_type=encoded_content_type,
68 headers=headers,
69 )
70
71
72@dataclass(frozen=True)
73class Target:
74 """Where a request goes: the parts of the path or URL it was given."""
75
76 scheme: str
77 host: str
78 port: str
79 path: str
80 query_string: str
81
82
83def split_target(path: str, *, query_params: dict[str, Any] | None = None) -> Target:
84 """Split a path, or a full http(s) URL, into where the request goes.
85
86 A path goes to the default scheme and host. A URL names its own; its
87 port is the one it gives, or its scheme's. A query string the path
88 carries comes first, and `query_params` go after it.
89 """
90 parts = urlsplit(str(path)) # path can be lazy
91
92 if parts.scheme in DEFAULT_PORTS:
93 scheme = parts.scheme
94 host = parts.hostname or DEFAULT_HOST
95 port = str(parts.port) if parts.port else DEFAULT_PORTS[scheme]
96 else:
97 scheme = DEFAULT_SCHEME
98 host = DEFAULT_HOST
99 port = DEFAULT_PORTS[scheme]
100
101 query_strings = [parts.query, urlencode(query_params or {}, doseq=True)]
102
103 return Target(
104 scheme=scheme,
105 host=host,
106 port=port,
107 path=parts.path,
108 query_string="&".join(part for part in query_strings if part),
109 )
110
111
112def build_encoded_request(
113 method: str,
114 target: Target,
115 *,
116 body: bytes,
117 content_type: str,
118 headers: dict[str, str] | None,
119) -> Request:
120 """Build a `Request` from a body that is already bytes."""
121 all_headers: dict[str, str] = dict(headers or {})
122
123 # Content headers follow the content type, not the byte count: a POST
124 # of an empty form still declares what it is, with Content-Length: 0,
125 # the same as a browser submitting a form with nothing filled in.
126 # Requests with no body source at all (a GET) resolve to no content
127 # type and get neither header.
128 if content_type:
129 all_headers["Content-Type"] = content_type
130 all_headers["Content-Length"] = str(len(body))
131
132 return Request(
133 method=method,
134 path=target.path,
135 headers=all_headers,
136 query_string=target.query_string,
137 body=body,
138 server_scheme=target.scheme,
139 server_name=target.host,
140 server_port=target.port,
141 remote_addr="127.0.0.1",
142 )
143
144
145def encode_request_body(
146 *,
147 form_data: dict[str, Any] | None,
148 json_data: Any,
149 body: bytes | str | None,
150 files: dict[str, Any] | None,
151 content_type: str | None,
152) -> tuple[bytes, str]:
153 """
154 Encode the body arguments into (bytes, content_type).
155
156 Exactly one body source may be given: form_data (optionally with files),
157 json_data, or a raw body. `content_type` only applies to a raw body.
158
159 A form encodes the way a browser would: urlencoded, and multipart only
160 when there are files. Views under test then see the content type they'd
161 see in production.
162 """
163 sources = [
164 form_data is not None or files is not None,
165 json_data is not None,
166 body is not None,
167 ]
168 if sum(sources) > 1:
169 raise TypeError(
170 "Pass only one of form_data/files, json_data, or body per request"
171 )
172 if content_type is not None and body is None:
173 if not any(sources):
174 # `post(path, content_type="application/json")` with nothing to
175 # send. Building an empty body here would hand the view a b"" that
176 # its content type says is parseable, and the failure would
177 # surface somewhere further in. Say it at the call instead.
178 raise TypeError(
179 "content_type needs a body — pass body=... alongside it "
180 '(body=b"" for a deliberately empty one)'
181 )
182 raise TypeError(
183 "content_type only applies to a raw body — form_data and json_data set their own"
184 )
185
186 if form_data is not None:
187 _refuse_files_in_form_data(form_data)
188
189 if json_data is not None:
190 return (
191 json.dumps(json_data, cls=PlainJSONEncoder).encode(),
192 "application/json",
193 )
194
195 if body is not None:
196 resolved_content_type = content_type or "application/octet-stream"
197 if isinstance(body, str):
198 # Encode a string body with the charset the content type
199 # declares, read the way `Request` reads it back, so the
200 # payload bytes match what the request advertises. Bytes pass
201 # through untouched.
202 _, content_params = parse_header_parameters(resolved_content_type)
203 body = body.encode(content_params.get("charset", "utf-8"))
204 return (body, resolved_content_type)
205
206 if files:
207 # Files can only travel as multipart, and any form fields sent with
208 # them ride along in the same body.
209 merged: dict[str, Any] = dict(form_data or {})
210 merged.update(files)
211 return (
212 encode_multipart(_BOUNDARY, merged),
213 _MULTIPART_CONTENT,
214 )
215
216 if form_data is not None or files is not None:
217 # A plain form — what a browser (and htmx) sends without a file input.
218 return (
219 urlencode(form_data or {}, doseq=True).encode(),
220 "application/x-www-form-urlencoded",
221 )
222
223 return (b"", "")
224
225
226def _refuse_files_in_form_data(form_data: dict[str, Any]) -> None:
227 """
228 A file in `form_data` would be sent as a text field holding the
229 object's `repr`, and the view would find no file. Say where it goes.
230 """
231 for name, value in form_data.items():
232 values = value if isinstance(value, list | tuple) else [value]
233 for one in values:
234 if _is_file_content(one):
235 raise TypeError(
236 f"form_data[{name!r}] is {type(one).__name__}, which is a "
237 "file's content. A form's text fields go in form_data and "
238 f"its files go in files: files={{{name!r}: ...}}"
239 )
240
241
242def _is_file_content(value: Any) -> bool:
243 if isinstance(value, bytes | bytearray | memoryview):
244 return True
245 # An open file, a BytesIO, an uploaded file: what can be read from.
246 return callable(getattr(value, "read", None))