char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somrearrange libs into lib and extensions into extsf4a0f5b

main
branchesmainall refs →
8.0 KiB250 linesraw
1use std::sync::{
2    Arc,
3    atomic::{AtomicBool, Ordering},
4};
5
6use avril_http::{Address, Engine};
7use bytes::Bytes;
8use tokio::io::{AsyncReadExt, AsyncWriteExt};
9
10#[tokio::test]
11async fn engine_streams_without_a_js_runtime() {
12    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
13    let (id, address) = engine
14        .listen(
15            Address::Tcp {
16                hostname: "127.0.0.1".into(),
17                port: 0,
18            },
19            Arc::new(|_| Ok(())),
20        )
21        .unwrap();
22    let Address::Tcp { hostname, port } = address else {
23        panic!("expected TCP");
24    };
25    let mut client = tokio::net::TcpStream::connect((hostname.as_str(), port))
26        .await
27        .unwrap();
28    client.write_all(b"POST /stream HTTP/1.1\r\nHost: example.test\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello").await.unwrap();
29    let metadata = engine.next(id).await.unwrap().unwrap();
30    assert_eq!(metadata.1, "POST");
31    assert_eq!(metadata.2, "http://example.test/stream");
32    let request = metadata.0;
33    let completed = engine.wait(request);
34    let mut body = Vec::new();
35    while let Some(bytes) = engine.read(request).await.unwrap() {
36        body.extend(bytes);
37    }
38    assert_eq!(body, b"hello");
39    assert!(engine.respond((request, 200, vec![], None, true)).unwrap());
40    assert!(
41        engine
42            .write(request, Some(Bytes::from_static(b"world")))
43            .await
44            .unwrap()
45    );
46    engine.write(request, None).await.unwrap();
47    let mut response = String::new();
48    client.read_to_string(&mut response).await.unwrap();
49    assert!(response.starts_with("HTTP/1.1 200"));
50    assert!(response.contains("world"));
51    assert!(!completed.await.unwrap());
52    engine.stop(id, false);
53    engine.finished(id).await.unwrap();
54}
55
56#[tokio::test]
57async fn native_file_ranges_preserve_keep_alive_framing() {
58    use avril_http::file::{header_pairs, headers, serve_file};
59
60    let directory = tempfile::tempdir().unwrap();
61    let path = directory.path().join("data.bin");
62    let bytes: Vec<u8> = (0..200_000).map(|i| (i % 251) as u8).collect();
63    std::fs::write(&path, &bytes).unwrap();
64    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
65    let (id, address) = engine
66        .listen(
67            Address::Tcp {
68                hostname: "127.0.0.1".into(),
69                port: 0,
70            },
71            Arc::new(|_| Ok(())),
72        )
73        .unwrap();
74    let Address::Tcp { hostname, port } = address else {
75        panic!("expected TCP");
76    };
77    let mut client = tokio::net::TcpStream::connect((hostname.as_str(), port))
78        .await
79        .unwrap();
80    client.write_all(b"GET /file HTTP/1.1\r\nHost: localhost\r\nRange: bytes=65530-131080\r\n\r\nGET /next HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").await.unwrap();
81    let receive = tokio::spawn(async move {
82        let mut bytes = Vec::new();
83        client.read_to_end(&mut bytes).await.unwrap();
84        bytes
85    });
86    let request = engine.next(id).await.unwrap().unwrap();
87    let complete = engine.wait(request.0);
88    let runtime = tokio::runtime::Handle::current();
89    let response = tokio::task::spawn_blocking(move || {
90        serve_file(
91            runtime,
92            &request.1,
93            headers(request.3).unwrap(),
94            "data.bin",
95            headers(vec![]).unwrap(),
96            || Ok((std::fs::File::open(&path)?, Arc::new(|| Ok(())))),
97        )
98    })
99    .await
100    .unwrap()
101    .unwrap();
102    engine
103        .respond_file(
104            request.0,
105            response.status().as_u16(),
106            header_pairs(response.headers()),
107            response.into_body().unwrap(),
108        )
109        .unwrap();
110    let next = engine.next(id).await.unwrap().unwrap().0;
111    let next_complete = engine.wait(next);
112    engine
113        .respond((next, 200, vec![], Some(Bytes::from_static(b"next")), false))
114        .unwrap();
115    let received = receive.await.unwrap();
116    let body_start = received
117        .windows(4)
118        .position(|part| part == b"\r\n\r\n")
119        .unwrap()
120        + 4;
121    assert!(received.starts_with(b"HTTP/1.1 206"));
122    assert_eq!(
123        &received[body_start..body_start + 65551],
124        &bytes[65530..=131080]
125    );
126    assert!(received[body_start + 65551..].starts_with(b"HTTP/1.1 200"));
127    assert!(received.ends_with(b"next"));
128    assert!(!complete.await.unwrap());
129    assert!(!next_complete.await.unwrap());
130    engine.stop(id, false);
131    engine.finished(id).await.unwrap();
132}
133
134#[tokio::test]
135async fn authorization_denials_do_not_leak_pending_responses() {
136    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
137    let allowed = Arc::new(AtomicBool::new(true));
138    let flag = allowed.clone();
139    let authorize = Arc::new(move |_: &Address| {
140        if flag.load(Ordering::Relaxed) {
141            Ok(())
142        } else {
143            Err("denied".into())
144        }
145    });
146    let (id, address) = engine
147        .listen(
148            Address::Tcp {
149                hostname: "127.0.0.1".into(),
150                port: 0,
151            },
152            authorize.clone(),
153        )
154        .unwrap();
155    let Address::Tcp { hostname, port } = address else {
156        panic!("expected TCP");
157    };
158    let mut client = tokio::net::TcpStream::connect((hostname.as_str(), port))
159        .await
160        .unwrap();
161    client
162        .write_all(b"GET / HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n")
163        .await
164        .unwrap();
165    let request = engine.next(id).await.unwrap().unwrap().0;
166    let completed = engine.wait(request);
167    allowed.store(false, Ordering::Relaxed);
168    assert_eq!(
169        engine
170            .respond((
171                request,
172                200,
173                vec![],
174                Some(Bytes::from_static(b"secret")),
175                false
176            ))
177            .unwrap_err()
178            .to_string(),
179        "denied"
180    );
181    assert!(
182        engine
183            .listen(
184                Address::Tcp {
185                    hostname: "127.0.0.1".into(),
186                    port: 0
187                },
188                authorize
189            )
190            .is_err()
191    );
192    engine.cancel(request);
193    let mut response = Vec::new();
194    client.read_to_end(&mut response).await.unwrap();
195    assert!(response.is_empty());
196    assert!(completed.await.unwrap());
197    engine.stop(id, false);
198    engine.finished(id).await.unwrap();
199}
200
201#[test]
202#[cfg(unix)]
203fn failed_listener_setup_leaves_no_socket_entry() {
204    let directory = tempfile::tempdir().unwrap();
205    let path = directory.path().join("failed.sock");
206    let runtime = tokio::runtime::Builder::new_current_thread()
207        .enable_all()
208        .build()
209        .unwrap();
210    let handle = runtime.handle().clone();
211    drop(runtime);
212    let engine = Arc::new(Engine::new(handle));
213    assert!(
214        engine
215            .listen(
216                Address::Unix {
217                    path: Some(path.to_str().unwrap().into())
218                },
219                Arc::new(|_| Ok(()))
220            )
221            .is_err()
222    );
223    assert!(!path.exists());
224}
225
226#[tokio::test]
227#[cfg(target_os = "linux")]
228async fn unix_cleanup_follows_the_pinned_directory_after_rename() {
229    let directory = tempfile::tempdir().unwrap();
230    let original = directory.path().join("original");
231    let moved = directory.path().join("moved");
232    std::fs::create_dir(&original).unwrap();
233    let path = original.join("server.sock");
234    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
235    let (id, _) = engine
236        .listen(
237            Address::Unix {
238                path: Some(path.to_str().unwrap().into()),
239            },
240            Arc::new(|_| Ok(())),
241        )
242        .unwrap();
243    std::fs::rename(&original, &moved).unwrap();
244    std::fs::create_dir(&original).unwrap();
245    std::fs::write(&path, "replacement").unwrap();
246    engine.stop(id, false);
247    engine.finished(id).await.unwrap();
248    assert!(!moved.join("server.sock").exists());
249    assert_eq!(std::fs::read_to_string(path).unwrap(), "replacement");
250}