Update test code to use Hyper 1.x and Axum instead of Warp (#178)

Author: sbstp

Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-authors = ["Simon Bernier St-Pierre <git@sbstp.ca>"]
+authors = ["Simon Bernier St-Pierre <git.sbstp.ca@gmail.com>"]
 edition = "2018"
 license = "MPL-2.0"
 name = "attohttpc"
@@ -38,24 +38,30 @@ rustls-opt-dep = { package = "rustls", version = "0.23.22", default-features = f
 serde = { version = "1.0.143", optional = true }
 serde_json = { version = "1.0.83", optional = true }
 serde_urlencoded = { version = "0.7.1", optional = true }
+tokio-rustls = "0.26.1"
 url = "2.2.2"
 webpki-roots = { version = "0.26.8", optional = true }
 
 [dev-dependencies]
 anyhow = "1.0.61"
+axum = { version = "0.8.1", features = ["multipart"] }
+axum-server = { version = "0.7.1", features = ["tls-rustls"] }
+bytes = "1.10.0"
 env_logger = "0.11.0"
-futures-util = { version = "0.3.23", default-features = false }
-http02 = { package = "http", version = "0.2" }
-hyper = "0.14.20"
+http-body-util = "0.1.2"
+hyper = { version = "1.6.0", features = ["full"] }
+hyper-util = "0.1.10"
 lazy_static = "1.4.0"
 multipart = { version = "0.18.0", default-features = false, features = [
     "server",
 ] }
+rustls-opt-dep = { package = "rustls", version = "0.23.22", default-features = false, features = [
+    "ring",
+    "std",
+    "tls12",
+] }
 rustls-pemfile = "2"
 tokio = { version = "1.20.1", features = ["full"] }
-tokio-rustls = "0.25.0"
-tokio-stream = { version = "0.1.9", features = ["net"] }
-warp = "0.3.2"
 
 [features]
 basic-auth = ["base64"]

tests/test_multipart.rs

@@ -1,102 +1,85 @@
-use std::io::{Cursor, Read};
-use std::net::SocketAddr;
-use std::sync::mpsc::{sync_channel, Receiver};
-use std::thread;
+use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
+use std::time::Duration;
 
-use mime::Mime;
-use multipart::server::Multipart;
-use tokio::runtime::Builder;
-use warp::Filter;
+use axum::extract::{DefaultBodyLimit, Multipart, State};
+use axum::routing::post;
+use axum::Router;
+use bytes::Bytes;
 
-fn start_server() -> (u16, Receiver<Option<String>>) {
+#[derive(Debug, PartialEq, Eq)]
+struct Part {
+    name: Option<String>,
+    file_name: Option<String>,
+    content_type: Option<String>,
+    data: Bytes,
+}
+
+async fn start_server() -> (u16, Receiver<Vec<Part>>) {
     let (send, recv) = sync_channel(1);
-    let rt = Builder::new_multi_thread().enable_io().enable_time().build().unwrap();
-    // ported from warp::multipart, which has a length limit (and we're generic over Read)
-    let filter = warp::path("multipart")
-        .and(
-            warp::header::<Mime>("content-type")
-                .and_then(|ct: Mime| async move {
-                    ct.get_param("boundary")
-                        .map(|mime| mime.to_string())
-                        .ok_or_else(warp::reject::reject)
-                })
-                .and(warp::body::bytes())
-                .map(|boundary, bytes| Multipart::with_body(Cursor::new(bytes), boundary)),
-        )
-        .map(move |mut form: Multipart<_>| {
-            let mut found_text = false;
-            let mut found_file = false;
-            let mut err = false;
-            let mut buf = String::new();
-            form.foreach_entry(|mut entry| {
-                if err {
-                    return;
-                }
-                entry.data.read_to_string(&mut buf).unwrap();
-                if !found_text && &*entry.headers.name == "Hello" && buf == "world!" {
-                    found_text = true;
-                } else if !found_file
-                    && &*entry.headers.name == "file"
-                    && entry.headers.filename.as_deref() == Some("hello.txt")
-                    && entry.headers.content_type.as_ref().map(|x| x.as_ref() == "text/plain") == Some(true)
-                    && buf == "Hello, world!"
-                {
-                    found_file = true;
-                } else {
-                    send.send(Some(format!("Unexpected entry {:?} = {:?}", entry.headers, buf)))
-                        .unwrap();
-                    err = true;
-                }
-                buf.clear();
-            })
-            .unwrap();
-            if err {
-                return "ERR";
-            }
-            send.send(Some(
-                match (found_text, found_file) {
-                    (false, false) => "Missing both fields!",
-                    (true, false) => "Missing file field!",
-                    (false, true) => "Missing text field!",
-                    (true, true) => {
-                        send.send(None).unwrap();
-                        return "OK";
-                    }
-                }
-                .to_string(),
-            ))
-            .unwrap();
-            "ERR"
-        });
-    let (addr, fut) =
-        rt.block_on(async { warp::serve(filter).bind_ephemeral("0.0.0.0:0".parse::<SocketAddr>().unwrap()) });
-    let port = addr.port();
-    thread::spawn(move || {
-        rt.block_on(fut);
+
+    async fn accept_form(State(send): State<SyncSender<Vec<Part>>>, mut multipart: Multipart) -> &'static str {
+        let mut parts = Vec::new();
+        while let Some(field) = multipart.next_field().await.unwrap() {
+            parts.push(Part {
+                name: field.name().map(|s| s.to_string()),
+                file_name: field.file_name().map(|s| s.to_string()),
+                content_type: field.content_type().map(|s| s.to_string()),
+                data: field.bytes().await.unwrap(),
+            });
+        }
+        send.send(parts).unwrap();
+        "OK"
+    }
+
+    let app = Router::new()
+        .route("/multipart", post(accept_form))
+        .layer(DefaultBodyLimit::disable())
+        .with_state(send);
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
     });
     (port, recv)
 }
 
-#[test]
-fn test_multipart_default() -> attohttpc::Result<()> {
-    let file = attohttpc::MultipartFile::new("file", b"Hello, world!")
+#[tokio::test(flavor = "multi_thread")]
+async fn test_multipart_default() -> attohttpc::Result<()> {
+    let file = attohttpc::MultipartFile::new("file", b"abc123")
         .with_type("text/plain")?
         .with_filename("hello.txt");
     let form = attohttpc::MultipartBuilder::new()
         .with_text("Hello", "world!")
         .with_file(file)
         .build()?;
 
-    let (port, recv) = start_server();
+    let (port, recv) = start_server().await;
 
     attohttpc::post(format!("http://localhost:{port}/multipart"))
         .body(form)
         .send()?
         .text()?;
 
-    if let Some(err) = recv.recv().unwrap() {
-        panic!("{}", err);
-    }
+    let parts = recv.recv_timeout(Duration::from_secs(5)).unwrap();
+    assert_eq!(parts.len(), 2);
+    assert_eq!(
+        parts,
+        vec![
+            Part {
+                name: Some("Hello".to_string()),
+                file_name: None,
+                content_type: None,
+                data: Bytes::from(&b"world!"[..])
+            },
+            Part {
+                name: Some("file".to_string()),
+                file_name: Some("hello.txt".to_string()),
+                content_type: Some("text/plain".to_string()),
+                data: Bytes::from(&b"abc123"[..])
+            }
+        ]
+    );
 
     Ok(())
 }

tests/test_proxy.rs

@@ -179,8 +179,6 @@ async fn test_http_url_with_https_proxy_refusal() -> Result<(), anyhow::Error> {
 #[cfg(any(feature = "tls-native", feature = "__rustls"))]
 #[tokio::test(flavor = "multi_thread")]
 async fn test_https_url_with_https_proxy_refusal() -> Result<(), anyhow::Error> {
-    // env_logger::init();
-
     let proxy_port = tools::start_refusing_proxy_server(true).await?;
     let proxy_url = Url::parse(&format!("https://localhost:{proxy_port}")).unwrap();
 

tests/test_redirection.rs

@@ -1,25 +1,37 @@
 use std::net::SocketAddr;
 
 use attohttpc::ErrorKind;
-use http02 as http;
-use tokio_stream::wrappers::TcpListenerStream;
-use warp::Filter;
+use axum::body::Body;
+use axum::response::Response;
+use axum::routing::get;
+use axum::Router;
+use http::StatusCode;
 
 async fn make_server() -> Result<u16, anyhow::Error> {
     let addr = SocketAddr::from(([127, 0, 0, 1], 0));
     let incoming = tokio::net::TcpListener::bind(&addr).await?;
     let local_addr = incoming.local_addr()?;
 
-    let a = warp::path("301").map(|| warp::redirect::redirect(http::Uri::from_static("/301")));
-    let b = warp::path("304").map(|| {
-        http::Response::builder()
-            .header("Location", "/304")
-            .status(http::StatusCode::NOT_MODIFIED)
-            .body("")
-    });
+    async fn x301() -> Response {
+        Response::builder()
+            .status(StatusCode::MOVED_PERMANENTLY)
+            .header("Location", "/301")
+            .body(Body::from(""))
+            .unwrap()
+    }
+
+    async fn x304() -> Response {
+        Response::builder()
+            .status(StatusCode::NOT_MODIFIED)
+            .body(Body::from(""))
+            .unwrap()
+    }
+
+    let app = Router::new().route("/301", get(x301)).route("/304", get(x304));
 
-    let server = warp::serve(a.or(b)).serve_incoming(TcpListenerStream::new(incoming));
-    tokio::spawn(server);
+    tokio::spawn(async move {
+        axum::serve(incoming, app).await.unwrap();
+    });
 
     Ok(local_addr.port())
 }

tests/tools/mod.rs

@@ -1,6 +1,5 @@
 mod proxy;
 mod servers;
-mod tls;
 
 pub use proxy::*;
 pub use servers::*;