Line data Source code
1 : //! The HTTP proxy through the routes the app calls, against a server that the tests answer by hand.
2 :
3 : use std::net::SocketAddr;
4 : use std::time::Duration;
5 :
6 : use http::{Request, Response};
7 :
8 : use crate::route_proxy_request;
9 : use crate::test::http_server_mock::spawn_http_server;
10 :
11 : const GUARD: Duration = Duration::from_secs(10);
12 :
13 54 : fn request(method: &str, path: &str, headers: &[(&str, &str)], body: &[u8]) -> Request<Vec<u8>> {
14 54 : let mut builder = Request::builder().method(method).uri(format!("dashql-native://localhost{path}"));
15 175 : for (name, value) in headers {
16 175 : builder = builder.header(*name, *value);
17 175 : }
18 54 : builder.body(body.to_vec()).unwrap()
19 54 : }
20 :
21 48 : fn header<'a>(response: &'a Response<Vec<u8>>, name: &str) -> Option<&'a str> {
22 48 : response.headers().get(name).map(|value| value.to_str().unwrap())
23 48 : }
24 :
25 14 : fn error_message(response: &Response<Vec<u8>>) -> String {
26 14 : assert_eq!(header(response, "dashql-error"), Some("true"), "{response:?}");
27 14 : let body: serde_json::Value = serde_json::from_slice(response.body()).unwrap();
28 14 : body["message"].as_str().unwrap().to_string()
29 14 : }
30 :
31 16 : async fn start_stream(address: SocketAddr, method: &str, path: &str, search: &str, extra: &[(&str, &str)], body: &[u8]) -> Response<Vec<u8>> {
32 16 : let endpoint = format!("http://{address}");
33 16 : let mut headers = vec![
34 16 : ("dashql-method", method),
35 16 : ("dashql-endpoint", endpoint.as_str()),
36 16 : ("dashql-path", path),
37 16 : ("dashql-search-params", search),
38 16 : ("dashql-read-timeout", "60000"),
39 : ];
40 16 : headers.extend_from_slice(extra);
41 16 : route_proxy_request(request("POST", "/http/streams", &headers, body)).await
42 16 : }
43 :
44 16 : fn stream_id(response: &Response<Vec<u8>>) -> String {
45 16 : assert_eq!(response.status(), 200, "{response:?}");
46 16 : header(response, "dashql-stream-id").unwrap().to_string()
47 16 : }
48 :
49 17 : async fn read_stream(id: &str, read_ms: &str, batch_ms: &str, bytes: &str) -> Response<Vec<u8>> {
50 17 : let headers = [("dashql-read-timeout", read_ms), ("dashql-batch-timeout", batch_ms), ("dashql-batch-bytes", bytes)];
51 17 : route_proxy_request(request("GET", &format!("/http/stream/{id}"), &headers, &[])).await
52 17 : }
53 :
54 : #[tokio::test]
55 1 : async fn answers_the_health_check() {
56 1 : let response = route_proxy_request(request("GET", "/health", &[], &[])).await;
57 1 : assert_eq!(response.status(), 200);
58 1 : assert_eq!(response.body(), br#"{"status":"ok"}"#);
59 1 : }
60 :
61 : #[tokio::test]
62 1 : async fn rejects_unknown_routes_and_methods() {
63 1 : assert_eq!(route_proxy_request(request("GET", "/unknown", &[], &[])).await.status(), 400);
64 1 : assert_eq!(route_proxy_request(request("POST", "/health", &[], &[])).await.status(), 400);
65 1 : assert_eq!(route_proxy_request(request("GET", "/http/streams", &[], &[])).await.status(), 404);
66 1 : assert_eq!(route_proxy_request(request("POST", "/http/stream/1", &[], &[])).await.status(), 404);
67 1 : }
68 :
69 : #[tokio::test]
70 1 : async fn requires_every_parameter_of_a_request() {
71 1 : let all = [
72 1 : ("dashql-method", "GET"),
73 1 : ("dashql-endpoint", "http://127.0.0.1:1"),
74 1 : ("dashql-path", "/"),
75 1 : ("dashql-search-params", ""),
76 1 : ("dashql-read-timeout", "10"),
77 1 : ];
78 5 : for missing in 0..all.len() {
79 25 : let headers: Vec<(&str, &str)> = all.iter().enumerate().filter(|(i, _)| *i != missing).map(|(_, h)| *h).collect();
80 5 : let response = route_proxy_request(request("POST", "/http/streams", &headers, &[])).await;
81 5 : assert_eq!(response.status(), 400);
82 5 : assert_eq!(error_message(&response), "header is missing", "without {:?}", all[missing].0);
83 1 : }
84 1 : }
85 :
86 : #[tokio::test]
87 1 : async fn rejects_invalid_parameters() {
88 1 : let headers = |method: &'static str, endpoint: &'static str, timeout: &'static str| [
89 3 : ("dashql-method", method),
90 3 : ("dashql-endpoint", endpoint),
91 3 : ("dashql-path", "/"),
92 3 : ("dashql-search-params", ""),
93 3 : ("dashql-read-timeout", timeout),
94 : ];
95 1 : let cases = [
96 1 : (headers("GET", "not a url", "10"), "header stores invalid endpoint"),
97 1 : (headers("BAD METHOD", "http://127.0.0.1:1", "10"), "header stores invalid method"),
98 1 : (headers("GET", "http://127.0.0.1:1", "soon"), "header is not an unsigned integer"),
99 1 : ];
100 3 : for (headers, message) in cases {
101 3 : let response = route_proxy_request(request("POST", "/http/streams", &headers, &[])).await;
102 3 : assert_eq!(error_message(&response), message);
103 1 : }
104 1 : }
105 :
106 : #[tokio::test]
107 1 : async fn sends_the_request_to_the_remote() {
108 1 : let (address, mut requests) = spawn_http_server().await;
109 1 : let start = start_stream(address, "POST", "/v1/query", "a=1&b=2", &[
110 1 : ("authorization", "Bearer token"),
111 1 : ("origin", "http://localhost:9002"),
112 1 : ("dashql-trace-id", "internal"),
113 1 : ], b"the body").await;
114 1 : stream_id(&start);
115 :
116 1 : let (received, _connection) = tokio::time::timeout(GUARD, requests.recv()).await.unwrap().unwrap();
117 1 : assert_eq!(received.line, "POST /v1/query?a=1&b=2 HTTP/1.1");
118 1 : assert_eq!(received.body, b"the body");
119 1 : assert_eq!(received.header("authorization"), Some("Bearer token"));
120 : // The fetch of the webview sets the origin, and our headers stop at the proxy
121 1 : assert_eq!(received.header("origin"), None);
122 4 : assert!(received.headers.iter().all(|(key, _)| !key.starts_with("dashql-")), "{:?}", received.headers);
123 1 : }
124 :
125 : #[tokio::test]
126 1 : async fn omits_an_empty_query() {
127 1 : let (address, mut requests) = spawn_http_server().await;
128 1 : stream_id(&start_stream(address, "GET", "/v1/info", "", &[], &[]).await);
129 1 : let (received, _connection) = tokio::time::timeout(GUARD, requests.recv()).await.unwrap().unwrap();
130 1 : assert_eq!(received.line, "GET /v1/info HTTP/1.1");
131 1 : }
132 :
133 : #[tokio::test]
134 1 : async fn streams_the_status_the_headers_and_the_body() {
135 1 : let (address, mut requests) = spawn_http_server().await;
136 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
137 1 : let (_, mut connection) = requests.recv().await.unwrap();
138 1 : connection.write_head("201 Created", &[("x-hyperdb-status", "ok")]).await;
139 1 : connection.write_chunk("hello ").await;
140 1 : connection.write_chunk("world").await;
141 1 : connection.finish().await;
142 :
143 1 : let response = read_stream(&id, "5000", "5000", "1000000").await;
144 1 : assert_eq!(response.status(), 201);
145 1 : assert_eq!(header(&response, "dashql-response-started"), Some("true"));
146 1 : assert_eq!(header(&response, "dashql-batch-event"), Some("StreamFinished"));
147 1 : assert_eq!(header(&response, "dashql-batch-bytes"), Some("11"));
148 1 : assert_eq!(header(&response, "x-hyperdb-status"), Some("ok"));
149 1 : assert_eq!(response.body(), b"hello world");
150 :
151 : // A finished stream is gone
152 1 : assert_eq!(read_stream(&id, "10", "10", "10").await.status(), 404);
153 1 : }
154 :
155 : #[tokio::test]
156 1 : async fn passes_an_error_of_the_remote_through_as_a_response() {
157 1 : let (address, mut requests) = spawn_http_server().await;
158 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
159 1 : let (_, mut connection) = requests.recv().await.unwrap();
160 1 : connection.write_head("503 Service Unavailable", &[]).await;
161 1 : connection.write_chunk("overloaded").await;
162 1 : connection.finish().await;
163 :
164 1 : let response = read_stream(&id, "5000", "5000", "1000000").await;
165 1 : assert_eq!(response.status(), 503);
166 1 : assert_eq!(header(&response, "dashql-error"), None);
167 1 : assert_eq!(response.body(), b"overloaded");
168 1 : }
169 :
170 : #[tokio::test]
171 1 : async fn flushes_a_batch_after_the_size_limit() {
172 1 : let (address, mut requests) = spawn_http_server().await;
173 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
174 1 : let (_, mut connection) = requests.recv().await.unwrap();
175 1 : connection.write_head("200 OK", &[]).await;
176 1 : connection.write_chunk("0123456789").await;
177 :
178 1 : let response = read_stream(&id, "5000", "5000", "5").await;
179 1 : assert_eq!(header(&response, "dashql-batch-event"), Some("FlushAfterBytes"));
180 1 : assert_eq!(response.body(), b"0123456789");
181 1 : assert_eq!(header(&response, "dashql-response-started"), Some("true"));
182 :
183 : // Later batches continue the body without another status
184 1 : connection.write_chunk("abc").await;
185 1 : connection.finish().await;
186 1 : let response = read_stream(&id, "5000", "5000", "1000").await;
187 1 : assert_eq!(header(&response, "dashql-batch-event"), Some("StreamFinished"));
188 1 : assert_eq!(header(&response, "dashql-response-started"), None);
189 1 : assert_eq!(response.body(), b"abc");
190 1 : }
191 :
192 : #[tokio::test]
193 1 : async fn flushes_a_batch_after_the_batch_timeout() {
194 1 : let (address, mut requests) = spawn_http_server().await;
195 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
196 1 : let (_, mut connection) = requests.recv().await.unwrap();
197 1 : connection.write_head("200 OK", &[]).await;
198 1 : connection.write_chunk("partial").await;
199 :
200 1 : let response = read_stream(&id, "5000", "100", "1000000").await;
201 1 : assert_eq!(header(&response, "dashql-batch-event"), Some("FlushAfterTimeout"));
202 1 : assert_eq!(response.body(), b"partial");
203 1 : }
204 :
205 : #[tokio::test]
206 1 : async fn reports_a_read_timeout_before_the_remote_answers() {
207 1 : let (address, mut requests) = spawn_http_server().await;
208 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
209 1 : let (_, _connection) = requests.recv().await.unwrap();
210 1 : let response = read_stream(&id, "50", "50", "1000").await;
211 1 : assert_eq!(error_message(&response), "reading from http stream timed out");
212 1 : }
213 :
214 : #[tokio::test]
215 1 : async fn keeps_the_status_when_the_body_is_slower_than_the_read_timeout() {
216 : // The remote answers, then computes for a while. The status and the headers must not get lost.
217 1 : let (address, mut requests) = spawn_http_server().await;
218 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
219 1 : let (_, mut connection) = requests.recv().await.unwrap();
220 1 : connection.write_head("503 Service Unavailable", &[("x-hyperdb-status", "busy")]).await;
221 :
222 1 : let first = read_stream(&id, "200", "50", "1000").await;
223 1 : assert_eq!(header(&first, "dashql-error"), None, "{first:?}");
224 1 : assert_eq!(first.status(), 503);
225 1 : assert_eq!(header(&first, "dashql-response-started"), Some("true"));
226 1 : assert_eq!(header(&first, "x-hyperdb-status"), Some("busy"));
227 1 : assert_eq!(header(&first, "dashql-batch-event"), Some("FlushAfterTimeout"));
228 :
229 1 : connection.write_chunk("late").await;
230 1 : connection.finish().await;
231 1 : let rest = read_stream(&id, "5000", "5000", "1000").await;
232 1 : assert_eq!(rest.body(), b"late");
233 1 : assert_eq!(header(&rest, "dashql-batch-event"), Some("StreamFinished"));
234 1 : }
235 :
236 : #[tokio::test]
237 1 : async fn the_remote_cannot_replace_the_headers_of_the_proxy() {
238 1 : let (address, mut requests) = spawn_http_server().await;
239 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
240 1 : let (_, mut connection) = requests.recv().await.unwrap();
241 1 : connection.write_head("200 OK", &[
242 1 : ("dashql-batch-event", "StreamFinished"),
243 1 : ("dashql-stream-id", "999"),
244 1 : ("dashql-error", "true"),
245 1 : ]).await;
246 1 : connection.write_chunk("0123456789").await;
247 :
248 1 : let response = read_stream(&id, "5000", "5000", "5").await;
249 1 : assert_eq!(header(&response, "dashql-batch-event"), Some("FlushAfterBytes"));
250 1 : assert_eq!(header(&response, "dashql-stream-id"), Some(id.as_str()));
251 1 : assert_eq!(header(&response, "dashql-error"), None);
252 1 : }
253 :
254 : #[tokio::test]
255 1 : async fn keeps_every_value_of_a_repeated_response_header() {
256 1 : let (address, mut requests) = spawn_http_server().await;
257 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
258 1 : let (_, mut connection) = requests.recv().await.unwrap();
259 1 : connection.write_head("200 OK", &[("set-cookie", "a=1"), ("set-cookie", "b=2")]).await;
260 1 : connection.finish().await;
261 :
262 1 : let response = read_stream(&id, "5000", "5000", "1000").await;
263 2 : let cookies: Vec<&str> = response.headers().get_all("set-cookie").iter().map(|value| value.to_str().unwrap()).collect();
264 1 : assert_eq!(cookies, vec!["a=1", "b=2"]);
265 1 : }
266 :
267 : #[tokio::test]
268 1 : async fn reports_a_remote_that_cannot_be_reached() {
269 : // Bind and close to get a port that nobody listens on
270 1 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
271 1 : let address = listener.local_addr().unwrap();
272 1 : drop(listener);
273 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
274 1 : let response = read_stream(&id, "5000", "5000", "1000").await;
275 1 : assert_eq!(error_message(&response), "http request failed");
276 : // The failed stream is gone
277 1 : assert_eq!(read_stream(&id, "10", "10", "10").await.status(), 404);
278 1 : }
279 :
280 : #[tokio::test]
281 1 : async fn reports_a_body_that_breaks_off() {
282 1 : let (address, mut requests) = spawn_http_server().await;
283 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
284 1 : let (_, mut connection) = requests.recv().await.unwrap();
285 1 : connection.write("HTTP/1.1 200 OK\r\ncontent-length: 100\r\n\r\nshort").await;
286 1 : drop(connection);
287 :
288 1 : let response = read_stream(&id, "5000", "5000", "1000").await;
289 1 : assert_eq!(error_message(&response), "reading chunk from http stream failed");
290 1 : assert_eq!(read_stream(&id, "10", "10", "10").await.status(), 404);
291 1 : }
292 :
293 : #[tokio::test]
294 1 : async fn requires_the_limits_of_a_read() {
295 3 : for headers in [
296 1 : vec![("dashql-batch-timeout", "1"), ("dashql-batch-bytes", "1")],
297 1 : vec![("dashql-read-timeout", "1"), ("dashql-batch-bytes", "1")],
298 1 : vec![("dashql-read-timeout", "1"), ("dashql-batch-timeout", "1")],
299 1 : ] {
300 3 : let response = route_proxy_request(request("GET", "/http/stream/0", &headers, &[])).await;
301 3 : assert_eq!(error_message(&response), "header is missing");
302 1 : }
303 1 : }
304 :
305 : #[tokio::test]
306 1 : async fn deleting_a_stream_ends_it() {
307 1 : let (address, mut requests) = spawn_http_server().await;
308 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
309 1 : let (_, _connection) = requests.recv().await.unwrap();
310 1 : let response = route_proxy_request(request("DELETE", &format!("/http/stream/{id}"), &[], &[])).await;
311 1 : assert_eq!(response.status(), 200);
312 1 : assert_eq!(read_stream(&id, "10", "10", "10").await.status(), 404);
313 : // Deleting twice is fine, the stream may have ended on its own
314 1 : assert_eq!(route_proxy_request(request("DELETE", &format!("/http/stream/{id}"), &[], &[])).await.status(), 200);
315 1 : }
316 :
317 : #[tokio::test]
318 1 : async fn deleting_a_stream_cancels_a_request_that_waits_for_the_remote() {
319 1 : let (address, mut requests) = spawn_http_server().await;
320 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
321 1 : let (_, mut connection) = requests.recv().await.unwrap();
322 1 : route_proxy_request(request("DELETE", &format!("/http/stream/{id}"), &[], &[])).await;
323 1 : tokio::time::timeout(GUARD, connection.closed()).await.expect("the proxy keeps the connection open");
324 1 : }
325 :
326 : #[tokio::test]
327 1 : async fn deleting_a_stream_cancels_a_body_in_progress() {
328 1 : let (address, mut requests) = spawn_http_server().await;
329 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
330 1 : let (_, mut connection) = requests.recv().await.unwrap();
331 1 : connection.write_head("200 OK", &[]).await;
332 1 : connection.write_chunk("partial").await;
333 1 : assert_eq!(read_stream(&id, "5000", "100", "1000000").await.body(), b"partial");
334 :
335 1 : route_proxy_request(request("DELETE", &format!("/http/stream/{id}"), &[], &[])).await;
336 1 : tokio::time::timeout(GUARD, connection.closed()).await.expect("the proxy keeps the connection open");
337 1 : }
338 :
339 : #[tokio::test]
340 1 : async fn deleting_a_stream_stops_a_reader_with_a_full_queue() {
341 1 : let (address, mut requests) = spawn_http_server().await;
342 1 : let id = stream_id(&start_stream(address, "GET", "/", "", &[], &[]).await);
343 1 : let (_, mut connection) = requests.recv().await.unwrap();
344 1 : connection.write_head("200 OK", &[]).await;
345 : // More chunks than the queue holds, nobody reads them
346 1 : for _ in 0..20 {
347 20 : connection.write_chunk("x").await;
348 20 : tokio::time::sleep(Duration::from_millis(5)).await;
349 : }
350 1 : route_proxy_request(request("DELETE", &format!("/http/stream/{id}"), &[], &[])).await;
351 1 : tokio::time::timeout(GUARD, connection.closed()).await.expect("the proxy keeps the connection open");
352 1 : }
|
