Allow package specific error codes in check_success (#3012)

Also:

- Added options to `Pipeline::send()`
- Propagate pipeline send options from `azure_core` to
`typespec_client_core`

Author: LarryOsterman

sdk/core/azure_core/benches/http_transport_benchmarks.rs

@@ -96,7 +96,7 @@ impl TestServiceClient {
 
         let response = self
             .pipeline
-            .send(&options.method_options.context, &mut request)
+            .send(&options.method_options.context, &mut request, None)
             .await?;
         if !response.status().is_success() {
             return Err(azure_core::Error::with_message(

sdk/core/azure_core/src/error/error_response.rs

@@ -108,19 +108,43 @@ struct ErrorDetailsInternal<'a> {
     message: Option<&'a str>,
 }
 
+/// Options for customizing the behavior of `check_success`.
+#[derive(Debug, Default)]
+pub struct CheckSuccessOptions {
+    /// A list of HTTP status codes that should be considered successful.
+    ///
+    /// If this list is empty, any 2xx status code is considered successful.
+    pub success_codes: &'static [u16],
+}
+
 /// Checks if the response is a success and if not, creates an appropriate error.
 ///
 /// # Arguments
 /// * `response` - The HTTP response to check.
+/// * `options` - Optional parameters to customize the success criteria.
 ///
 /// # Returns
 /// * `Ok(RawResponse)` if the response is a success.
 /// * `Err(Error)` if the response is an error, with details extracted from the response
 ///   body if possible.
 ///
-pub async fn check_success(response: BufResponse) -> crate::Result<BufResponse> {
+pub async fn check_success(
+    response: BufResponse,
+    options: Option<CheckSuccessOptions>,
+) -> crate::Result<BufResponse> {
     let status = response.status();
-    if status.is_success() {
+
+    if options
+        .as_ref()
+        .map(|o| {
+            if o.success_codes.is_empty() {
+                status.is_success()
+            } else {
+                o.success_codes.contains(&status)
+            }
+        })
+        .unwrap_or_else(|| status.is_success())
+    {
         return Ok(response);
     }
 
@@ -186,7 +210,32 @@ mod tests {
             Bytes::from_static(br#"{"error": {"code":"teapot","message":"I'm a teapot"}}"#),
         );
 
-        let err = check_success(response).await.unwrap_err();
+        let err = check_success(response, None).await.unwrap_err();
+        let kind = err.kind();
+        assert!(matches!(
+            kind,
+            ErrorKind::HttpResponse {
+                status: StatusCode::ImATeapot,
+                error_code,
+                raw_response: None
+            }
+            if error_code.as_deref() == Some("teapot")
+        ));
+    }
+
+    #[tokio::test]
+    async fn matching_against_custom_http_error_empty_set() {
+        let mut headers = Headers::new();
+        headers.insert(headers::CONTENT_TYPE, "application/json".to_string());
+        let response = BufResponse::from_bytes(
+            StatusCode::ImATeapot,
+            headers,
+            Bytes::from_static(br#"{"error": {"code":"teapot","message":"I'm a teapot"}}"#),
+        );
+
+        let err = check_success(response, Some(CheckSuccessOptions { success_codes: &[] }))
+            .await
+            .unwrap_err();
         let kind = err.kind();
         assert!(matches!(
             kind,
@@ -199,13 +248,63 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn matching_against_custom_http_error_in_set() {
+        let mut headers = Headers::new();
+        headers.insert(headers::CONTENT_TYPE, "application/json".to_string());
+        let response = BufResponse::from_bytes(
+            StatusCode::ImATeapot,
+            headers,
+            Bytes::from_static(br#"{"error": {"code":"teapot","message":"I'm a teapot"}}"#),
+        );
+
+        let _ = check_success(
+            response,
+            Some(CheckSuccessOptions {
+                success_codes: &[418],
+            }),
+        )
+        .await
+        .expect("Should be a success return");
+    }
+
+    #[tokio::test]
+    async fn matching_against_custom_http_error_in_set_success_should_fail() {
+        let mut headers = Headers::new();
+        headers.insert(headers::CONTENT_TYPE, "application/json".to_string());
+        let response = BufResponse::from_bytes(
+            StatusCode::Ok,
+            headers,
+            Bytes::from_static(br#"{"error": {"code":"teapot","message":"I'm a teapot"}}"#),
+        );
+
+        let err = check_success(
+            response,
+            Some(CheckSuccessOptions {
+                success_codes: &[418],
+            }),
+        )
+        .await
+        .expect_err("Should be a failure return");
+        let kind = err.kind();
+        assert!(matches!(
+            kind,
+            ErrorKind::HttpResponse {
+                status: StatusCode::Ok,
+                error_code,
+                raw_response: None
+            }
+            if error_code.as_deref() == Some("teapot")
+        ));
+    }
+
     #[tokio::test]
     async fn matching_against_http_error_no_body() {
         let mut headers = Headers::new();
         headers.insert(headers::ERROR_CODE, "testError".to_string());
         let response = BufResponse::from_bytes(StatusCode::ImATeapot, headers, Bytes::new());
 
-        let err = check_success(response).await.unwrap_err();
+        let err = check_success(response, None).await.unwrap_err();
         let kind = err.kind();
         assert_eq!(
             *kind,
@@ -227,7 +326,7 @@ mod tests {
             Bytes::from_static(br#"{"json": "error"}"#),
         );
 
-        let err = check_success(response).await.unwrap_err();
+        let err = check_success(response, None).await.unwrap_err();
         let kind = err.kind();
         assert_eq!(
             *kind,

sdk/core/azure_core/src/http/pager.rs

@@ -223,7 +223,7 @@ impl<P: Page> ItemIterator<P> {
     ///                 .append_pair("api-version", &api_version);
     ///         }
     ///         let resp = pipeline
-    ///           .send(&Context::new(), &mut req)
+    ///           .send(&Context::new(), &mut req, None)
     ///           .await?;
     ///         let (status, headers, body) = resp.deconstruct();
     ///         let bytes = body.collect().await?;
@@ -269,7 +269,7 @@ impl<P: Page> ItemIterator<P> {
     ///             req.insert_header("x-ms-continuation", continuation);
     ///         }
     ///         let resp: Response<ListItemsResult> = pipeline
-    ///           .send(&Context::new(), &mut req)
+    ///           .send(&Context::new(), &mut req, None)
     ///           .await?
     ///           .into();
     ///         Ok(PagerResult::from_response_header(resp, &HeaderName::from_static("x-next-continuation")))
@@ -416,7 +416,7 @@ impl<P> PageIterator<P> {
     ///                 .append_pair("api-version", &api_version);
     ///         }
     ///         let resp = pipeline
-    ///           .send(&Context::new(), &mut req)
+    ///           .send(&Context::new(), &mut req, None)
     ///           .await?;
     ///         let (status, headers, body) = resp.deconstruct();
     ///         let bytes = body.collect().await?;
@@ -453,7 +453,7 @@ impl<P> PageIterator<P> {
     ///             req.insert_header("x-ms-continuation", continuation);
     ///         }
     ///         let resp: Response<ListItemsResult> = pipeline
-    ///           .send(&Context::new(), &mut req)
+    ///           .send(&Context::new(), &mut req, None)
     ///           .await?
     ///           .into();
     ///         Ok(PagerResult::from_response_header(resp, &HeaderName::from_static("x-ms-continuation")))

sdk/core/azure_core/src/http/pipeline.rs

@@ -2,16 +2,19 @@
 // Licensed under the MIT License.
 
 use super::policies::ClientRequestIdPolicy;
-use crate::http::{
-    headers::{RETRY_AFTER_MS, X_MS_RETRY_AFTER_MS},
-    policies::{
-        Policy, PublicApiInstrumentationPolicy, RequestInstrumentationPolicy, UserAgentPolicy,
+use crate::{
+    error::CheckSuccessOptions,
+    http::{
+        check_success,
+        headers::{RETRY_AFTER_MS, X_MS_RETRY_AFTER_MS},
+        policies::{
+            Policy, PublicApiInstrumentationPolicy, RequestInstrumentationPolicy, UserAgentPolicy,
+        },
+        ClientOptions,
     },
-    ClientOptions,
 };
 use std::{
     any::{Any, TypeId},
-    ops::Deref,
     sync::Arc,
 };
 use typespec_client_core::http::{
@@ -40,6 +43,37 @@ use typespec_client_core::http::{
 #[derive(Debug, Clone)]
 pub struct Pipeline(http::Pipeline);
 
+/// Options for sending a request through the pipeline.
+#[derive(Debug, Default)]
+pub struct PipelineSendOptions {
+    /// If true, skip all checks including `[check_success]`.
+    pub skip_checks: bool,
+
+    /// Options for `[check_success]`. If `skip_checks` is true, this field is ignored.
+    pub check_success: CheckSuccessOptions,
+}
+
+/// Internal structure used to pass options to the core pipeline.
+#[derive(Debug, Default)]
+struct CorePipelineSendOptions {
+    check_success: CheckSuccessOptions,
+    skip_checks: bool,
+}
+
+impl PipelineSendOptions {
+    /// Deconstructs the `PipelineSendOptions` into its core components.
+    #[expect(private_interfaces)]
+    pub fn deconstruct(self) -> (CorePipelineSendOptions, Option<http::PipelineSendOptions>) {
+        (
+            CorePipelineSendOptions {
+                skip_checks: self.skip_checks,
+                check_success: self.check_success,
+            },
+            None,
+        )
+    }
+}
+
 impl Pipeline {
     /// Creates a new pipeline given the client library crate name and version,
     /// alone with user-specified and client library-specified policies.
@@ -116,6 +150,32 @@ impl Pipeline {
             Some(pipeline_options),
         ))
     }
+
+    /// Sends the request through the pipeline, returning the response or an error.
+    ///
+    /// # Arguments
+    /// * `ctx` - The context for the request.
+    /// * `request` - The request to send.
+    /// * `options` - Options for sending the request, including check success options. If none, `[check_success]` will not be called.
+    ///
+    /// # Returns
+    /// A [`http::BufResponse`] if the request was successful, or an `Error` if it failed.
+    /// If the response status code indicates an HTTP error, the function will attempt to parse the error response
+    /// body into an `ErrorResponse` and include it in the `Error`.
+    pub async fn send(
+        &self,
+        ctx: &http::Context<'_>,
+        request: &mut http::Request,
+        options: Option<PipelineSendOptions>,
+    ) -> crate::Result<http::BufResponse> {
+        let (core_send_options, send_options) = options.unwrap_or_default().deconstruct();
+        let result = self.0.send(ctx, request, send_options).await?;
+        if !core_send_options.skip_checks {
+            check_success(result, Some(core_send_options.check_success)).await
+        } else {
+            Ok(result)
+        }
+    }
 }
 
 #[inline]
@@ -125,13 +185,13 @@ fn push_unique<T: Policy + 'static>(policies: &mut Vec<Arc<dyn Policy>>, policy:
     }
 }
 
-// TODO: Should we instead use the newtype pattern?
-impl Deref for Pipeline {
-    type Target = http::Pipeline;
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
+// // TODO: Should we instead use the newtype pattern?
+// impl Deref for Pipeline {
+//     type Target = http::Pipeline;
+//     fn deref(&self) -> &Self::Target {
+//         &self.0
+//     }
+// }
 
 #[cfg(test)]
 mod tests {
@@ -205,7 +265,7 @@ mod tests {
 
         // Act
         pipeline
-            .send(&ctx, &mut request)
+            .send(&ctx, &mut request, None)
             .await
             .expect("Pipeline execution failed");
     }
@@ -259,7 +319,7 @@ mod tests {
 
         // Act
         pipeline
-            .send(&ctx, &mut request)
+            .send(&ctx, &mut request, None)
             .await
             .expect("Pipeline execution failed");
     }
@@ -313,7 +373,7 @@ mod tests {
 
         // Act
         pipeline
-            .send(&ctx, &mut request)
+            .send(&ctx, &mut request, None)
             .await
             .expect("Pipeline execution failed");
     }
@@ -374,7 +434,7 @@ mod tests {
 
         // Act
         pipeline
-            .send(&ctx, &mut request)
+            .send(&ctx, &mut request, None)
             .await
             .expect("Pipeline execution failed");
     }

sdk/core/azure_core/src/http/poller.rs

@@ -292,7 +292,7 @@ where
     ///             .append_pair("api-version", &api_version);
     ///
     ///         let resp = pipeline
-    ///             .send(&Context::new(), &mut req)
+    ///             .send(&Context::new(), &mut req, None)
     ///             .await?;
     ///         let (status, headers, body) = resp.deconstruct();
     ///         let bytes = body.collect().await?;

sdk/core/azure_core_opentelemetry/tests/telemetry_service_implementation.rs

@@ -131,7 +131,7 @@ impl TestServiceClient {
 
         let response = self
             .pipeline
-            .send(&options.method_options.context, &mut request)
+            .send(&options.method_options.context, &mut request, None)
             .await?;
         if !response.status().is_success() {
             return Err(azure_core::Error::with_message(

sdk/core/azure_core_opentelemetry/tests/telemetry_service_macros.rs

@@ -106,7 +106,7 @@ impl TestServiceClientWithMacros {
 
         let response = self
             .pipeline
-            .send(&options.method_options.context, &mut request)
+            .send(&options.method_options.context, &mut request, None)
             .await?;
         if !response.status().is_success() {
             return Err(azure_core::Error::with_message(
@@ -157,7 +157,7 @@ impl TestServiceClientWithMacros {
 
         let response = self
             .pipeline
-            .send(&options.method_options.context, &mut request)
+            .send(&options.method_options.context, &mut request, None)
             .await?;
         if !response.status().is_success() {
             return Err(azure_core::Error::with_message(