test chunked encoding

Author: bdbai

nyquest-backend-tests/src/fixtures.rs

@@ -1,2 +1,3 @@
 mod requests;
 mod responses;
+mod scenarios;

nyquest-backend-tests/src/fixtures/responses.rs

@@ -87,7 +87,7 @@ mod tests {
         const PATH: &str = "responses/status_codes";
         const STATUS_CODES: [u16; 4] = [400, 404, 500, 502];
         let _handle = crate::add_hyper_fixture(PATH, |mut req| async move {
-            let mut res = Response::default();
+            let mut res = Response::<Full<Bytes>>::default();
             let body = req.body_mut().collect().await.ok().and_then(|bytes| {
                 let status = String::from_utf8_lossy(&bytes.to_bytes()).parse().ok()?;
                 StatusCode::from_u16(status).ok()
@@ -140,7 +140,7 @@ mod tests {
         const HEADER_NAME: &str = "X-Test-Header";
         const HEADER_VALUE: &str = "test-value";
         let _handle = crate::add_hyper_fixture(PATH, |_req| async move {
-            let mut res = Response::default();
+            let mut res = Response::<Full<Bytes>>::default();
             res.headers_mut()
                 .insert(HEADER_NAME, HEADER_VALUE.parse().unwrap());
             (res, Ok(()))

nyquest-backend-tests/src/fixtures/scenarios.rs

@@ -0,0 +1 @@
+mod chunked_encoding;

nyquest-backend-tests/src/fixtures/scenarios/chunked_encoding.rs

@@ -0,0 +1,58 @@
+#[cfg(test)]
+mod tests {
+    use futures::stream::{self};
+    use http_body_util::BodyExt;
+    use hyper::{Method, Response};
+    use nyquest::Request as NyquestRequest;
+
+    use crate::*;
+
+    #[test]
+    fn test_chunked_encoding() {
+        const PATH: &str = "scenarios/chunked_encoding";
+        const CHUNKS: [&str; 4] = ["Hello", ", ", "chunked ", "world!"];
+
+        let _handle = crate::add_hyper_fixture(PATH, {
+            move |req| {
+                async move {
+                    // Create a streaming response body by yielding each chunk with a small delay
+                    let stream = stream::iter(CHUNKS.iter().map(|chunk| {
+                        let chunk = Bytes::copy_from_slice(chunk.as_bytes());
+                        Ok::<_, hyper::Error>(hyper::body::Frame::data(chunk))
+                    }));
+
+                    // Convert the stream to a boxed body
+                    let body = http_body_util::StreamBody::new(stream).boxed();
+                    let res = Response::new(body);
+
+                    (res, (req.method() == Method::GET).then_some(()).ok_or(req))
+                }
+            }
+        });
+
+        let expected_content = CHUNKS.concat();
+        let builder = crate::init_builder_blocking().unwrap();
+
+        let assertions = |content: &str| {
+            assert_eq!(content, expected_content);
+        };
+
+        #[cfg(feature = "blocking")]
+        {
+            let client = builder.clone().build_blocking().unwrap();
+            let res = client.request(NyquestRequest::get(PATH)).unwrap();
+            let content = res.text().unwrap();
+            assertions(&content);
+        }
+
+        #[cfg(feature = "async")]
+        {
+            let content = TOKIO_RT.block_on(async {
+                let client = builder.build_async().await.unwrap();
+                let res = client.request(NyquestRequest::get(PATH)).await.unwrap();
+                res.text().await.unwrap()
+            });
+            assertions(&content);
+        }
+    }
+}

nyquest-backend-tests/src/lib.rs

@@ -10,7 +10,7 @@ use std::{
     sync::{LazyLock, Mutex, Once},
 };
 
-use http_body_util::Full;
+use http_body_util::{BodyExt, Full};
 use hyper::{
     body::{self, Bytes},
     server::conn::http1,
@@ -45,7 +45,28 @@ impl Drop for HyperFixtureHandle {
     }
 }
 
-type FixtureAssertionResult = (Response<Full<Bytes>>, Result<(), Request<body::Incoming>>);
+// BoxedBody for supporting streaming/chunked responses
+type BoxedBody = http_body_util::combinators::BoxBody<Bytes, hyper::Error>;
+
+// New type allowing both Full<Bytes> and BoxedBody response types
+// This makes existing tests compatible with the new changes
+type FixtureAssertionResult = (ResponseWrapper, Result<(), Request<body::Incoming>>);
+
+struct ResponseWrapper(Response<BoxedBody>);
+
+impl From<Response<Full<Bytes>>> for ResponseWrapper {
+    fn from(resp: Response<Full<Bytes>>) -> Self {
+        let resp = resp.map(|body| body.map_err(|_| unreachable!()).boxed());
+        ResponseWrapper(resp)
+    }
+}
+
+impl From<Response<BoxedBody>> for ResponseWrapper {
+    fn from(resp: Response<BoxedBody>) -> Self {
+        ResponseWrapper(resp)
+    }
+}
+
 struct HyperServiceFixture {
     svc: Box<
         dyn Fn(
@@ -56,20 +77,28 @@ struct HyperServiceFixture {
     >,
     assertion_failed_request: Option<Request<body::Incoming>>,
 }
+
 static HYPER_SERVICE_FIXTURES: Mutex<BTreeMap<String, HyperServiceFixture>> =
     Mutex::new(BTreeMap::new());
 
-fn add_hyper_fixture<Fut: Future<Output = FixtureAssertionResult> + Send + 'static>(
+fn add_hyper_fixture<Fut, Resp>(
     url: impl Into<String>,
     svc_fn: impl Fn(Request<body::Incoming>) -> Fut + Send + Sync + 'static,
-) -> HyperFixtureHandle {
+) -> HyperFixtureHandle
+where
+    Fut: Future<Output = (Resp, Result<(), Request<body::Incoming>>)> + Send + 'static,
+    Resp: Into<ResponseWrapper>,
+{
     let mut url: String = url.into();
     if !url.starts_with('/') {
         url.insert(0, '/');
     }
     let svc = Box::new(move |req| {
         let fut = svc_fn(req);
-        Box::pin(async move { fut.await }) as _
+        Box::pin(async move {
+            let (resp, result) = fut.await;
+            (resp.into(), result)
+        }) as _
     });
     let fixture = HyperServiceFixture {
         svc,
@@ -83,20 +112,22 @@ fn add_hyper_fixture<Fut: Future<Output = FixtureAssertionResult> + Send + 'stat
     HyperFixtureHandle(url)
 }
 
-async fn handle_service(req: Request<body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
+async fn handle_service(req: Request<body::Incoming>) -> Result<Response<BoxedBody>, Infallible> {
     let path = req.uri().path().to_owned();
     let fut = {
         let services = HYPER_SERVICE_FIXTURES.lock().unwrap();
         let fixture = services.get(&*path).unwrap();
         (fixture.svc)(req)
     };
     let (response, result) = fut.await;
+
     if let Err(req) = result {
         let mut services = HYPER_SERVICE_FIXTURES.lock().unwrap();
         let fixture = services.get_mut(&*path).unwrap();
         fixture.assertion_failed_request = Some(req);
     }
-    Ok(response)
+
+    Ok(response.0)
 }
 
 async fn setup_hyper_impl() -> Result<String, io::Error> {