refactor(app/integration): forward-compatible test code (#3673)

see linkerd/linkerd2#8733 for more
information.

see #3559 and
#3614 for more information
on the `ForwardCompatibleBody<B>` wrapper.

this branch updates test code in `linkerd-app-integration` so that it
interacts with request and response bodies via an adapter that polls for
frames in a manner consistent with the 1.0 api of `http_body`.

this allows us to limit the diff in
#3504, which will only
need to remove this adapter once using hyper 1.0.

see #3671 and #3672, which perform the same change for
`linkerd-app-inbound` and `linkerd-app-outbound`, respectively.

---

* chore(app/integration): `linkerd-http-body-compat` test dependency

Signed-off-by: katelyn martin <[email protected]>

* refactor(app/integration): generalize `hyper::Body`

Signed-off-by: katelyn martin <[email protected]>

* refactor(app/integration): use `ForwardCompatibleBody`

Signed-off-by: katelyn martin <[email protected]>

---------

Signed-off-by: katelyn martin <[email protected]>

Author: cratelyn

Cargo.lock

@@ -1481,6 +1481,7 @@ dependencies = [
  "linkerd-app-admin",
  "linkerd-app-core",
  "linkerd-app-test",
+ "linkerd-http-body-compat",
  "linkerd-meshtls",
  "linkerd-metrics",
  "linkerd-tracing",

linkerd/app/integration/Cargo.toml

@@ -63,6 +63,7 @@ flate2 = { version = "1", default-features = false, features = [
 ] }
 # Log streaming isn't enabled by default globally, but we want to test it.
 linkerd-app-admin = { path = "../admin", features = ["log-streaming"] }
+linkerd-http-body-compat = { path = "../../http/body-compat" }
 # No code from this crate is actually used; only necessary to enable the Rustls
 # implementation.
 linkerd-meshtls = { path = "../../meshtls", features = ["rustls"] }

linkerd/app/integration/src/tests/profiles.rs

@@ -682,7 +682,9 @@ mod grpc_retry {
         assert_eq!(res.status(), 200);
         assert_eq!(res.headers().get(&GRPC_STATUS), None);
 
-        let mut body = res.into_body();
+        let mut body = res
+            .map(linkerd_http_body_compat::ForwardCompatibleBody::new)
+            .into_body();
         let trailers = trailers(&mut body).await;
         assert_eq!(trailers.get(&GRPC_STATUS), Some(&GRPC_STATUS_OK));
         assert_eq!(retries.load(Ordering::Relaxed), 2);
@@ -726,7 +728,9 @@ mod grpc_retry {
         assert_eq!(res.status(), 200);
         assert_eq!(res.headers().get(&GRPC_STATUS), None);
 
-        let mut body = res.into_body();
+        let mut body = res
+            .map(linkerd_http_body_compat::ForwardCompatibleBody::new)
+            .into_body();
 
         let data = data(&mut body).await;
         assert_eq!(data, Bytes::from("hello world"));
@@ -777,7 +781,9 @@ mod grpc_retry {
         assert_eq!(res.status(), 200);
         assert_eq!(res.headers().get(&GRPC_STATUS), None);
 
-        let mut body = res.into_body();
+        let mut body = res
+            .map(linkerd_http_body_compat::ForwardCompatibleBody::new)
+            .into_body();
 
         let frame1 = data(&mut body).await;
         assert_eq!(frame1, Bytes::from("hello"));
@@ -790,21 +796,38 @@ mod grpc_retry {
         assert_eq!(retries.load(Ordering::Relaxed), 1);
     }
 
-    async fn data(body: &mut hyper::Body) -> Bytes {
+    async fn data<B>(body: &mut linkerd_http_body_compat::ForwardCompatibleBody<B>) -> B::Data
+    where
+        B: http_body::Body + Unpin,
+        B::Data: std::fmt::Debug,
+        B::Error: std::fmt::Debug,
+    {
         let data = body
-            .data()
+            .frame()
             .await
-            .expect("body data frame must not be eaten")
-            .unwrap();
+            .expect("a result")
+            .expect("a frame")
+            .into_data()
+            .expect("a chunk of data");
         tracing::info!(?data);
         data
     }
-    async fn trailers(body: &mut hyper::Body) -> http::HeaderMap {
+
+    async fn trailers<B>(
+        body: &mut linkerd_http_body_compat::ForwardCompatibleBody<B>,
+    ) -> http::HeaderMap
+    where
+        B: http_body::Body + Unpin,
+        B::Error: std::fmt::Debug,
+    {
         let trailers = body
-            .trailers()
+            .frame()
             .await
-            .expect("trailers future should not fail")
-            .expect("response should have trailers");
+            .expect("a result")
+            .expect("a frame")
+            .into_trailers()
+            .ok()
+            .expect("a trailers frame");
         tracing::info!(?trailers);
         trailers
     }