kube_client/client/

mod.rs

//! API client for interacting with the Kubernetes API
//!
//! The [`Client`] uses standard kube error handling.
//!
//! This client can be used on its own or in conjunction with the [`Api`][crate::api::Api]
//! type for more structured interaction with the kubernetes API.
//!
//! The [`Client`] can also be used with [`Discovery`](crate::Discovery) to dynamically
//! retrieve the resources served by the kubernetes API.
use either::{Either, Left, Right};
use futures::{AsyncBufRead, StreamExt, TryStream, TryStreamExt, future::BoxFuture};
use http::{self, Request, Response};
use http_body_util::BodyExt;
#[cfg(feature = "ws")]
use hyper_util::rt::TokioIo;
use jiff::Timestamp;
use k8s_openapi::apimachinery::pkg::apis::meta::v1 as k8s_meta_v1;
use kube_core::{discovery::v2::ACCEPT_AGGREGATED_DISCOVERY_V2, response::Status};
use serde::de::DeserializeOwned;
use serde_json::{self, Value};
#[cfg(feature = "ws")]
use tokio_tungstenite::{WebSocketStream, tungstenite as ws};
use tokio_util::{
    codec::{FramedRead, LinesCodec, LinesCodecError},
    io::StreamReader,
};
use tower::{BoxError, Service, ServiceExt as _, buffer::Buffer};
use tower_http::ServiceExt as _;

pub use self::body::Body;
use crate::{Config, Error, Result, api::WatchEvent, config::Kubeconfig};

mod auth;
mod body;
mod builder;
pub use kube_core::discovery::v2::{
    APIGroupDiscovery, APIGroupDiscoveryList, APIResourceDiscovery, APISubresourceDiscovery,
    APIVersionDiscovery, GroupVersionKind as DiscoveryGroupVersionKind,
};
#[cfg_attr(docsrs, doc(cfg(feature = "unstable-client")))]
#[cfg(feature = "unstable-client")]
mod client_ext;
#[cfg_attr(docsrs, doc(cfg(feature = "unstable-client")))]
#[cfg(feature = "unstable-client")]
pub use client_ext::scope;
mod config_ext;
pub use auth::Error as AuthError;
pub use config_ext::ConfigExt;
pub mod middleware;
pub mod retry;

#[cfg(any(feature = "rustls-tls", feature = "openssl-tls"))]
mod tls;

#[cfg(feature = "openssl-tls")]
pub use tls::openssl_tls::Error as OpensslTlsError;
#[cfg(feature = "rustls-tls")]
pub use tls::rustls_tls::Error as RustlsTlsError;
#[cfg(feature = "ws")]
mod upgrade;

#[cfg(feature = "oauth")]
#[cfg_attr(docsrs, doc(cfg(feature = "oauth")))]
pub use auth::OAuthError;

#[cfg(feature = "oidc")]
#[cfg_attr(docsrs, doc(cfg(feature = "oidc")))]
pub use auth::oidc_errors;

#[cfg(feature = "ws")]
pub use upgrade::UpgradeConnectionError;

#[cfg(feature = "kubelet-debug")]
#[cfg_attr(docsrs, doc(cfg(feature = "kubelet-debug")))]
mod kubelet_debug;

pub use builder::{ClientBuilder, DynBody};

/// Client for connecting with a Kubernetes cluster.
///
/// The easiest way to instantiate the client is either by
/// inferring the configuration from the environment using
/// [`Client::try_default`] or with an existing [`Config`]
/// using [`Client::try_from`].
#[cfg_attr(docsrs, doc(cfg(feature = "client")))]
#[derive(Clone)]
pub struct Client {
    // - `Buffer` for cheap clone
    // - `BoxFuture` for dynamic response future type
    inner: Buffer<Request<Body>, BoxFuture<'static, Result<Response<Body>, BoxError>>>,
    default_ns: String,
    valid_until: Option<Timestamp>,
}

/// Represents a WebSocket connection.
/// Value returned by [`Client::connect`].
#[cfg(feature = "ws")]
#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
pub struct Connection {
    stream: WebSocketStream<TokioIo<hyper::upgrade::Upgraded>>,
    protocol: upgrade::StreamProtocol,
}

#[cfg(feature = "ws")]
#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
impl Connection {
    /// Return true if the stream supports graceful close signaling.
    pub fn supports_stream_close(&self) -> bool {
        self.protocol.supports_stream_close()
    }

    /// Transform into the raw WebSocketStream.
    pub fn into_stream(self) -> WebSocketStream<TokioIo<hyper::upgrade::Upgraded>> {
        self.stream
    }
}

/// Constructors and low-level api interfaces.
///
/// Most users only need [`Client::try_default`] or [`Client::new`] from this block.
///
/// The many various lower level interfaces here are for more advanced use-cases with specific requirements.
impl Client {
    /// Create a [`Client`] using a custom `Service` stack.
    ///
    /// [`ConfigExt`](crate::client::ConfigExt) provides extensions for
    /// building a custom stack.
    ///
    /// To create with the default stack with a [`Config`], use
    /// [`Client::try_from`].
    ///
    /// To create with the default stack with an inferred [`Config`], use
    /// [`Client::try_default`].
    ///
    /// # Example
    ///
    /// ```rust
    /// # async fn doc() -> Result<(), Box<dyn std::error::Error>> {
    /// use kube::{client::ConfigExt, Client, Config};
    /// use tower::{BoxError, ServiceBuilder};
    /// use hyper_util::rt::TokioExecutor;
    ///
    /// let config = Config::infer().await?;
    /// let service = ServiceBuilder::new()
    ///     .layer(config.base_uri_layer())
    ///     .option_layer(config.auth_layer()?)
    ///     .map_err(BoxError::from)
    ///     .service(hyper_util::client::legacy::Client::builder(TokioExecutor::new()).build_http());
    /// let client = Client::new(service, config.default_namespace);
    /// # Ok(())
    /// # }
    /// ```
    pub fn new<S, B, T>(service: S, default_namespace: T) -> Self
    where
        S: Service<Request<Body>, Response = Response<B>> + Send + 'static,
        S::Future: Send + 'static,
        S::Error: Into<BoxError>,
        B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
        B::Error: Into<BoxError>,
        T: Into<String>,
    {
        // Transform response body to `crate::client::Body` and use type erased error to avoid type parameters.
        let service = service
            .map_response_body(Body::wrap_body)
            .map_err(Into::into)
            .boxed();
        Self {
            inner: Buffer::new(service, 1024),
            default_ns: default_namespace.into(),
            valid_until: None,
        }
    }

    /// Sets an expiration timestamp to the client, which has to be checked by the user using [`Client::valid_until`] function.
    pub fn with_valid_until(self, valid_until: Option<Timestamp>) -> Self {
        Client { valid_until, ..self }
    }

    /// Get the expiration timestamp of the client, if it has been set.
    pub fn valid_until(&self) -> &Option<Timestamp> {
        &self.valid_until
    }

    /// Create and initialize a [`Client`] using the inferred configuration.
    ///
    /// Will use [`Config::infer`] which attempts to load the local kubeconfig first,
    /// and then if that fails, trying the in-cluster environment variables.
    ///
    /// Will fail if neither configuration could be loaded.
    ///
    /// ```rust
    /// # async fn doc() -> Result<(), Box<dyn std::error::Error>> {
    /// # use kube::Client;
    /// let client = Client::try_default().await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// If you already have a [`Config`] then use [`Client::try_from`](Self::try_from)
    /// instead.
    pub async fn try_default() -> Result<Self> {
        Self::try_from(Config::infer().await.map_err(Error::InferConfig)?)
    }

    /// Get the default namespace for the client
    ///
    /// The namespace is either configured on `context` in the kubeconfig,
    /// falls back to `default` when running locally,
    /// or uses the service account's namespace when deployed in-cluster.
    pub fn default_namespace(&self) -> &str {
        &self.default_ns
    }

    /// Perform a raw HTTP request against the API and return the raw response back.
    /// This method can be used to get raw access to the API which may be used to, for example,
    /// create a proxy server or application-level gateway between localhost and the API server.
    pub async fn send(&self, request: Request<Body>) -> Result<Response<Body>> {
        let mut svc = self.inner.clone();
        let res = svc
            .ready()
            .await
            .map_err(Error::Service)?
            .call(request)
            .await
            .map_err(|err| {
                // Error decorating request
                err.downcast::<Error>()
                    .map(|e| *e)
                    // Error requesting
                    .or_else(|err| err.downcast::<hyper::Error>().map(|err| Error::HyperError(*err)))
                    // Error from another middleware
                    .unwrap_or_else(Error::Service)
            })?;
        Ok(res)
    }

    /// Make WebSocket connection.
    #[cfg(feature = "ws")]
    #[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
    pub async fn connect(&self, request: Request<Vec<u8>>) -> Result<Connection> {
        use http::header::HeaderValue;
        let (mut parts, body) = request.into_parts();
        parts
            .headers
            .insert(http::header::CONNECTION, HeaderValue::from_static("Upgrade"));
        parts
            .headers
            .insert(http::header::UPGRADE, HeaderValue::from_static("websocket"));
        parts.headers.insert(
            http::header::SEC_WEBSOCKET_VERSION,
            HeaderValue::from_static("13"),
        );
        let key = tokio_tungstenite::tungstenite::handshake::client::generate_key();
        parts.headers.insert(
            http::header::SEC_WEBSOCKET_KEY,
            key.parse().expect("valid header value"),
        );
        upgrade::StreamProtocol::add_to_headers(&mut parts.headers)?;

        let res = self.send(Request::from_parts(parts, Body::from(body))).await?;
        // 4xx/5xx bodies are `metav1.Status`. Surface them as `Error::Api`
        // before the upgrade check, which only understands 101.
        let res = handle_api_errors(res).await?;
        let protocol = upgrade::verify_response(&res, &key).map_err(Error::UpgradeConnection)?;
        match hyper::upgrade::on(res).await {
            Ok(upgraded) => Ok(Connection {
                stream: WebSocketStream::from_raw_socket(
                    TokioIo::new(upgraded),
                    ws::protocol::Role::Client,
                    None,
                )
                .await,
                protocol,
            }),

            Err(e) => Err(Error::UpgradeConnection(
                UpgradeConnectionError::GetPendingUpgrade(e),
            )),
        }
    }

    /// Perform a raw HTTP request against the API and deserialize the response
    /// as JSON to some known type.
    pub async fn request<T>(&self, request: Request<Vec<u8>>) -> Result<T>
    where
        T: DeserializeOwned,
    {
        let text = self.request_text(request).await?;

        serde_json::from_str(&text).map_err(|e| {
            tracing::warn!("{}, {:?}", text, e);
            Error::SerdeError(e)
        })
    }

    /// Perform a raw HTTP request against the API and get back the response
    /// as a string
    pub async fn request_text(&self, request: Request<Vec<u8>>) -> Result<String> {
        let res = self.send(request.map(Body::from)).await?;
        let res = handle_api_errors(res).await?;
        let body_bytes = res.into_body().collect().await?.to_bytes();
        let text = String::from_utf8(body_bytes.to_vec()).map_err(Error::FromUtf8)?;
        Ok(text)
    }

    /// Perform a raw HTTP request against the API and stream the response body.
    ///
    /// The response can be processed using [`AsyncReadExt`](futures::AsyncReadExt)
    /// and [`AsyncBufReadExt`](futures::AsyncBufReadExt).
    pub async fn request_stream(&self, request: Request<Vec<u8>>) -> Result<impl AsyncBufRead + use<>> {
        let res = self.send(request.map(Body::from)).await?;
        let res = handle_api_errors(res).await?;
        // Map the error, since we want to convert this into an `AsyncBufReader` using
        // `into_async_read` which specifies `std::io::Error` as the stream's error type.
        let body = res.into_body().into_data_stream().map_err(std::io::Error::other);
        Ok(body.into_async_read())
    }

    /// Perform a raw HTTP request against the API and get back either an object
    /// deserialized as JSON or a [`Status`] Object.
    pub async fn request_status<T>(&self, request: Request<Vec<u8>>) -> Result<Either<T, Status>>
    where
        T: DeserializeOwned,
    {
        let text = self.request_text(request).await?;
        // It needs to be JSON:
        let v: Value = serde_json::from_str(&text).map_err(Error::SerdeError)?;
        if v["kind"] == "Status" {
            tracing::trace!("Status from {}", text);
            Ok(Right(serde_json::from_str::<Status>(&text).map_err(|e| {
                tracing::warn!("{}, {:?}", text, e);
                Error::SerdeError(e)
            })?))
        } else {
            Ok(Left(serde_json::from_str::<T>(&text).map_err(|e| {
                tracing::warn!("{}, {:?}", text, e);
                Error::SerdeError(e)
            })?))
        }
    }

    /// Perform a raw request and get back a stream of [`WatchEvent`] objects
    pub async fn request_events<T>(
        &self,
        mut request: Request<Vec<u8>>,
    ) -> Result<impl TryStream<Item = Result<WatchEvent<T>>> + use<T>>
    where
        T: Clone + DeserializeOwned,
    {
        // Kubernetes >=1.37 gzips watch streams one member per event, but our decompression
        // middleware only decodes the first member, so opt watch streams out of compression.
        // Drop this once tower-http decodes multi-member gzip again: tower-rs/tower-http#737,
        // fixed upstream by Nullus157/async-compression#489.
        request
            .headers_mut()
            .entry(http::header::ACCEPT_ENCODING)
            .or_insert(http::HeaderValue::from_static("identity"));
        let res = self.send(request.map(Body::from)).await?;
        let res = handle_api_errors(res).await?;
        // trace!("Streaming from {} -> {}", res.url(), res.status().as_str());
        tracing::trace!("headers: {:?}", res.headers());

        let frames = FramedRead::new(
            StreamReader::new(res.into_body().into_data_stream().map_err(|e| {
                // Resumable watch termination. TLS stacks nest an `io::Error`
                // of kind `UnexpectedEof`. Chunk decoders still report
                // "unexpected EOF during chunk" in the error text. Classify
                // either case by keeping the original error as the source.
                let chunk_eof = e.to_string().contains("unexpected EOF during chunk");
                if error_chain_contains_unexpected_eof(&e) || chunk_eof {
                    return std::io::Error::new(std::io::ErrorKind::UnexpectedEof, e);
                }
                std::io::Error::other(e)
            })),
            LinesCodec::new(),
        );

        Ok(frames.filter_map(|res| async {
            match res {
                Ok(line) => match serde_json::from_str::<WatchEvent<T>>(&line) {
                    Ok(event) => Some(Ok(event)),
                    Err(e) => {
                        // Ignore EOF error that can happen for incomplete line from `decode_eof`.
                        if e.is_eof() {
                            return None;
                        }

                        // Got general error response
                        if let Ok(status) = serde_json::from_str::<Status>(&line) {
                            return Some(Err(Error::Api(status.boxed())));
                        }
                        // Parsing error
                        Some(Err(Error::SerdeError(e)))
                    }
                },

                Err(LinesCodecError::Io(e)) => match e.kind() {
                    // Client timeout
                    std::io::ErrorKind::TimedOut => {
                        tracing::warn!("timeout in poll: {}", e); // our client timeout
                        None
                    }
                    // Unexpected EOF from the chunked decoder, or a nested
                    // `io::ErrorKind::UnexpectedEof`. Tends to happen after 300+s
                    // of watching. End the stream so it can be resumed.
                    std::io::ErrorKind::UnexpectedEof => {
                        tracing::warn!("eof in poll: {:?}", e);
                        None
                    }
                    _ => Some(Err(Error::ReadEvents(e))),
                },

                // Reached the maximum line length without finding a newline.
                // This should never happen because we're using the default `usize::MAX`.
                Err(LinesCodecError::MaxLineLengthExceeded) => {
                    Some(Err(Error::LinesCodecMaxLineLengthExceeded))
                }
            }
        }))
    }
}

/// Low level discovery methods using `k8s_openapi` types.
///
/// Consider using the [`discovery`](crate::discovery) module for
/// easier-to-use variants of this functionality.
/// The following methods might be deprecated to avoid confusion between similarly named types within `discovery`.
impl Client {
    /// Returns apiserver version.
    pub async fn apiserver_version(&self) -> Result<k8s_openapi::apimachinery::pkg::version::Info> {
        self.request(Request::get("/version").body(vec![]).map_err(Error::HttpError)?)
            .await
    }

    /// Lists api groups that apiserver serves.
    pub async fn list_api_groups(&self) -> Result<k8s_meta_v1::APIGroupList> {
        self.request(Request::get("/apis").body(vec![]).map_err(Error::HttpError)?)
            .await
    }

    /// Lists resources served in given API group.
    ///
    /// ### Example usage:
    /// ```rust
    /// # async fn scope(client: kube::Client) -> Result<(), Box<dyn std::error::Error>> {
    /// let apigroups = client.list_api_groups().await?;
    /// for g in apigroups.groups {
    ///     let ver = g
    ///         .preferred_version
    ///         .as_ref()
    ///         .or_else(|| g.versions.first())
    ///         .expect("preferred or versions exists");
    ///     let apis = client.list_api_group_resources(&ver.group_version).await?;
    ///     dbg!(apis);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn list_api_group_resources(&self, apiversion: &str) -> Result<k8s_meta_v1::APIResourceList> {
        let url = format!("/apis/{apiversion}");
        self.request(Request::get(url).body(vec![]).map_err(Error::HttpError)?)
            .await
    }

    /// Lists versions of `core` a.k.a. `""` legacy API group.
    pub async fn list_core_api_versions(&self) -> Result<k8s_meta_v1::APIVersions> {
        self.request(Request::get("/api").body(vec![]).map_err(Error::HttpError)?)
            .await
    }

    /// Lists resources served in particular `core` group version.
    pub async fn list_core_api_resources(&self, version: &str) -> Result<k8s_meta_v1::APIResourceList> {
        let url = format!("/api/{version}");
        self.request(Request::get(url).body(vec![]).map_err(Error::HttpError)?)
            .await
    }
}

/// Aggregated Discovery API methods
///
/// These methods use the Aggregated Discovery API (available since Kubernetes 1.26, stable in 1.30)
/// to fetch all API resources in a single request, reducing the number of API calls compared to
/// the traditional discovery methods.
impl Client {
    /// Returns aggregated discovery for all API groups served at /apis.
    ///
    /// This uses the Aggregated Discovery API to fetch all non-core API groups
    /// and their resources in a single request.
    ///
    /// Requires Kubernetes 1.26+ (beta) or 1.30+ (stable).
    ///
    /// ### Example usage:
    /// ```rust,no_run
    /// # async fn scope(client: kube::Client) -> Result<(), Box<dyn std::error::Error>> {
    /// let discovery = client.list_api_groups_aggregated().await?;
    /// for group in discovery.items {
    ///     let name = group.metadata.as_ref().and_then(|m| m.name.as_ref());
    ///     println!("Group: {:?}", name);
    ///     for version in group.versions {
    ///         println!("  Version: {:?}", version.version);
    ///         for resource in version.resources {
    ///             println!("    Resource: {:?}", resource.resource);
    ///         }
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn list_api_groups_aggregated(&self) -> Result<APIGroupDiscoveryList> {
        self.request(
            Request::get("/apis")
                .header(http::header::ACCEPT, ACCEPT_AGGREGATED_DISCOVERY_V2)
                .body(vec![])
                .map_err(Error::HttpError)?,
        )
        .await
    }

    /// Returns aggregated discovery for core API group served at /api.
    ///
    /// This uses the Aggregated Discovery API to fetch the core API group
    /// and all its resources in a single request.
    ///
    /// Requires Kubernetes 1.26+ (beta) or 1.30+ (stable).
    ///
    /// ### Example usage:
    /// ```rust,no_run
    /// # async fn scope(client: kube::Client) -> Result<(), Box<dyn std::error::Error>> {
    /// let discovery = client.list_core_api_versions_aggregated().await?;
    /// for group in discovery.items {
    ///     for version in group.versions {
    ///         println!("Core version: {:?}", version.version);
    ///         for resource in version.resources {
    ///             println!("  Resource: {:?} (scope: {:?})", resource.resource, resource.scope);
    ///         }
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn list_core_api_versions_aggregated(&self) -> Result<APIGroupDiscoveryList> {
        self.request(
            Request::get("/api")
                .header(http::header::ACCEPT, ACCEPT_AGGREGATED_DISCOVERY_V2)
                .body(vec![])
                .map_err(Error::HttpError)?,
        )
        .await
    }
}

// `err` or an error reached through `std::error::Error::source` is an
// `std::io::Error` of kind `UnexpectedEof`.
fn error_chain_contains_unexpected_eof(err: &(dyn std::error::Error + 'static)) -> bool {
    // HTTP and TLS stacks nest a handful of errors. Stop if a custom `source`
    // implementation cycles so a watch poll cannot stall.
    const MAX_SOURCE_DEPTH: usize = 16;
    let mut current = Some(err);
    for _ in 0..MAX_SOURCE_DEPTH {
        let Some(err) = current else {
            return false;
        };
        if let Some(io_err) = err.downcast_ref::<std::io::Error>()
            && io_err.kind() == std::io::ErrorKind::UnexpectedEof
        {
            return true;
        }
        current = err.source();
    }
    false
}

/// Kubernetes returned error handling
///
/// Either kube returned an explicit ApiError struct,
/// or it somehow returned something we couldn't parse as one.
///
/// In either case, present an ApiError upstream.
async fn handle_api_errors(res: Response<Body>) -> Result<Response<Body>> {
    let status = res.status();
    if status.is_client_error() || status.is_server_error() {
        // trace!("Status = {:?} for {}", status, res.url());
        let body_bytes = res.into_body().collect().await?.to_bytes();
        let text = String::from_utf8(body_bytes.to_vec()).map_err(Error::FromUtf8)?;
        // Print better debug when things do fail
        // trace!("Parsing error: {}", text);
        if let Ok(status) = serde_json::from_str::<Status>(&text) {
            tracing::debug!("Unsuccessful: {status:?}");
            Err(Error::Api(status.boxed()))
        } else {
            // Not every error response is a JSON `Status`. A proxy, ingress
            // controller, or a bare API path that doesn't exist (e.g. a CRD
            // that isn't installed yet) can return a plain-text or HTML body
            // instead. This is routine rather than exceptional, so it's
            // logged at `debug` like the parsed case above rather than `warn`.
            let text = text.trim();
            tracing::debug!("Unsuccessful data error parse: {text}");
            let status = Status::failure(text, "Failed to parse error data").with_code(status.as_u16());
            tracing::debug!("Unsuccessful: {status:?} (reconstruct)");
            Err(Error::Api(status.boxed()))
        }
    } else {
        Ok(res)
    }
}

impl TryFrom<Config> for Client {
    type Error = Error;

    /// Builds a default [`Client`] from a [`Config`].
    ///
    /// See [`ClientBuilder`] or [`Client::new`] if more customization is required
    fn try_from(config: Config) -> Result<Self> {
        Ok(ClientBuilder::try_from(config)?.build())
    }
}

impl TryFrom<Kubeconfig> for Client {
    type Error = Error;

    fn try_from(kubeconfig: Kubeconfig) -> Result<Self> {
        let config = Config::try_from(kubeconfig)?;
        Ok(ClientBuilder::try_from(config)?.build())
    }
}

#[cfg(test)]
mod tests {
    use std::{fmt, io::ErrorKind, pin::pin, time::Duration};

    use crate::{
        Api, Client, Error, Result,
        api::WatchEvent,
        client::Body,
        config::{AuthInfo, Cluster, Context, Kubeconfig, NamedAuthInfo, NamedCluster, NamedContext},
    };

    use bytes::Bytes;
    use futures::{Stream, StreamExt};
    use http::{Request, Response};
    use http_body::Frame;
    use http_body_util::StreamBody;
    use k8s_openapi::api::core::v1::Pod;
    use kube_core::metadata::PartialObjectMeta;
    use tower_test::mock;

    const WATCH_POLL_TIMEOUT: Duration = Duration::from_secs(2);

    #[tokio::test]
    async fn test_default_ns() {
        let (mock_service, _) = mock::pair::<Request<Body>, Response<Body>>();
        let client = Client::new(mock_service, "test-namespace");
        assert_eq!(client.default_namespace(), "test-namespace");
    }

    #[tokio::test]
    async fn test_try_from_kubeconfig() {
        let config = Kubeconfig {
            current_context: Some("test-context".to_string()),
            auth_infos: vec![NamedAuthInfo {
                name: "test-user".to_string(),
                auth_info: Some(AuthInfo::default()), // <-- empty but valid
                ..Default::default()
            }],
            contexts: vec![NamedContext {
                name: "test-context".to_string(),
                context: Some(Context {
                    cluster: "test-cluster".to_string(),
                    user: Some("test-user".to_string()),
                    namespace: Some("test-namespace".to_string()),
                    ..Default::default()
                }),
                ..Default::default()
            }],
            clusters: vec![NamedCluster {
                name: "test-cluster".to_string(),
                cluster: Some(Cluster {
                    server: Some("http://localhost:8080".to_string()),
                    ..Default::default()
                }),
                ..Default::default()
            }],
            ..Default::default()
        };
        let client = Client::try_from(config).expect("Failed to create client from kubeconfig");
        assert_eq!(client.default_namespace(), "test-namespace");
    }

    #[tokio::test]
    async fn test_mock() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            // Receive a request for pod and respond with some data
            let mut handle = pin!(handle);
            let (request, send) = handle.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::GET);
            assert_eq!(request.uri().to_string(), "/api/v1/namespaces/default/pods/test");
            let pod: Pod = serde_json::from_value(serde_json::json!({
                "apiVersion": "v1",
                "kind": "Pod",
                "metadata": {
                    "name": "test",
                    "annotations": { "kube-rs": "test" },
                },
                "spec": {
                    "containers": [{ "name": "test", "image": "test-image" }],
                }
            }))
            .unwrap();
            send.send_response(Response::new(Body::from(serde_json::to_vec(&pod).unwrap())));
        });

        let pods: Api<Pod> = Api::default_namespaced(Client::new(mock_service, "default"));
        let pod = pods.get("test").await.unwrap();
        assert_eq!(pod.metadata.annotations.unwrap().get("kube-rs").unwrap(), "test");
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn test_partial_object_meta_get_uses_metadata_header() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (request, send) = handle.next_request().await.expect("service not called");
            // Verify the metadata-only Accept header is set
            assert_eq!(
                request.headers().get(http::header::ACCEPT).unwrap(),
                "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1"
            );
            let pod_meta: PartialObjectMeta<Pod> =
                serde_json::from_value(serde_json::json!({
                    "apiVersion": "meta.k8s.io/v1",
                    "kind": "PartialObjectMetadata",
                    "metadata": {
                        "name": "test",
                    },
                }))
                .unwrap();
            send.send_response(Response::new(Body::from(
                serde_json::to_vec(&pod_meta).unwrap(),
            )));
        });

        let pods: Api<PartialObjectMeta<Pod>> =
            Api::default_namespaced(Client::new(mock_service, "default"));
        let pod = pods.get("test").await.unwrap();
        assert_eq!(pod.metadata.name, Some("test".to_string()));
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn test_partial_object_meta_list_uses_metadata_header() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (request, send) = handle.next_request().await.expect("service not called");
            assert_eq!(
                request.headers().get(http::header::ACCEPT).unwrap(),
                "application/json;as=PartialObjectMetadataList;g=meta.k8s.io;v=v1"
            );
            send.send_response(Response::new(Body::from(
                serde_json::to_vec(&serde_json::json!({
                    "apiVersion": "meta.k8s.io/v1",
                    "kind": "PartialObjectMetadataList",
                    "metadata": { "resourceVersion": "1" },
                    "items": [],
                }))
                .unwrap(),
            )));
        });

        let pods: Api<PartialObjectMeta<Pod>> =
            Api::default_namespaced(Client::new(mock_service, "default"));
        let list = pods.list(&Default::default()).await.unwrap();
        assert_eq!(list.items.len(), 0);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn test_partial_object_meta_patch_uses_metadata_header() {
        use kube_core::params::{Patch, PatchParams};

        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (request, send) = handle.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::PATCH);
            assert_eq!(
                request.headers().get(http::header::ACCEPT).unwrap(),
                "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1"
            );
            send.send_response(Response::new(Body::from(
                serde_json::to_vec(&serde_json::json!({
                    "apiVersion": "meta.k8s.io/v1",
                    "kind": "PartialObjectMetadata",
                    "metadata": { "name": "test" },
                }))
                .unwrap(),
            )));
        });

        let pods: Api<PartialObjectMeta<Pod>> =
            Api::default_namespaced(Client::new(mock_service, "default"));
        let patch = Patch::Merge(serde_json::json!({ "metadata": { "labels": { "a": "b" } } }));
        let pod = pods.patch("test", &PatchParams::default(), &patch).await.unwrap();
        assert_eq!(pod.metadata.name, Some("test".to_string()));
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn test_partial_object_meta_watch_uses_metadata_header() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (request, send) = handle.next_request().await.expect("service not called");
            assert_eq!(
                request.headers().get(http::header::ACCEPT).unwrap(),
                "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1"
            );
            // Send an empty response to close the stream
            send.send_response(Response::new(Body::from(vec![])));
        });

        let pods: Api<PartialObjectMeta<Pod>> =
            Api::default_namespaced(Client::new(mock_service, "default"));
        let _ = pods.watch(&Default::default(), "0").await;
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn test_watch_error_response_is_surfaced() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (_request, send) = handle.next_request().await.expect("service not called");
            // apiservers pretty-print error bodies for some clients, so this can span several lines
            let body = serde_json::to_vec_pretty(&serde_json::json!({
                "kind": "Status",
                "apiVersion": "v1",
                "metadata": {},
                "status": "Failure",
                "message": "pods is forbidden",
                "reason": "Forbidden",
                "code": 403,
            }))
            .unwrap();
            send.send_response(
                Response::builder()
                    .status(http::StatusCode::FORBIDDEN)
                    .body(Body::from(body))
                    .unwrap(),
            );
        });

        let pods: Api<Pod> = Api::default_namespaced(Client::new(mock_service, "default"));
        let Err(err) = pods.watch(&Default::default(), "0").await else {
            panic!("watch with an error response should fail");
        };
        assert!(matches!(&err, Error::Api(s) if s.code == 403), "got {err:?}");
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn test_non_json_error_response_is_reconstructed() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (_request, send) = handle.next_request().await.expect("service not called");
            // Some routers in front of the apiserver (or a probe for a CRD
            // that isn't installed yet) return a plain-text 404 instead of a
            // JSON `Status`.
            send.send_response(
                Response::builder()
                    .status(http::StatusCode::NOT_FOUND)
                    .body(Body::from(b"404 page not found\n".to_vec()))
                    .unwrap(),
            );
        });

        let pods: Api<Pod> = Api::default_namespaced(Client::new(mock_service, "default"));
        let Err(err) = pods.get("test").await else {
            panic!("get with a non-JSON error response should fail");
        };
        let Error::Api(status) = &err else {
            panic!("expected Error::Api, got {err:?}");
        };
        assert_eq!(status.code, 404);
        assert_eq!(status.reason, "Failed to parse error data");
        assert_eq!(status.message, "404 page not found");
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_direct_unexpected_eof_delivers_added_then_ends() {
        // Neutral wording: classification must use `ErrorKind`, not the
        // rustls missing-close-notify message.
        let body = body_from_chunks(vec![
            Ok(pod_event_frame("ADDED", "11")),
            Err(std::io::Error::new(ErrorKind::UnexpectedEof, "connection closed")),
        ]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "ADDED", "11");
        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_wrapped_unexpected_eof_delivers_modified_then_ends() {
        let body = body_from_chunks(vec![
            Ok(pod_event_frame("MODIFIED", "22")),
            Err(nested_io(
                &["outer transport interruption", "inner transport interruption"],
                std::io::Error::new(ErrorKind::UnexpectedEof, "connection closed"),
            )),
        ]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "MODIFIED", "22");
        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_unexpected_eof_before_events_ends() {
        let body = body_from_chunks(vec![Err(nested_io(
            &["outer transport interruption", "inner transport interruption"],
            std::io::Error::new(ErrorKind::UnexpectedEof, "connection closed"),
        ))]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_unexpected_eof_after_truncated_line_keeps_resource_version() {
        let mut payload = pod_event_json("ADDED", "33");
        payload.push('\n');
        payload.push_str(r#"{"type":"MODIFIED","object":{"apiVersion":"v1""#);
        let body = body_from_chunks(vec![
            Ok(Bytes::from(payload)),
            Err(std::io::Error::new(ErrorKind::UnexpectedEof, "connection closed")),
        ]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "ADDED", "33");
        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_clean_end_after_truncated_line_keeps_resource_version() {
        let mut payload = pod_event_json("MODIFIED", "34");
        payload.push('\n');
        payload.push_str(r#"{"type":"ADDED","object":{"apiVersion":"v1""#);
        let body = body_from_chunks::<std::io::Error>(vec![Ok(Bytes::from(payload))]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "MODIFIED", "34");
        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_clean_termination_delivers_every_event() {
        let payload = format!("{}{}", pod_event_frame_str("ADDED", "1"), pod_event_frame_str("MODIFIED", "2"));
        let body = body_from_chunks::<std::io::Error>(vec![Ok(Bytes::from(payload))]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "ADDED", "1");
        expect_pod(next_event(&mut stream).await, "MODIFIED", "2");
        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_chunk_decoder_eof_text_ends_stream() {
        let body = body_from_chunks(vec![
            Ok(pod_event_frame("ADDED", "44")),
            Err(ChunkDecoderEof),
        ]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "ADDED", "44");
        expect_end(next_event(&mut stream).await);
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_malformed_json_is_serde_error() {
        let body = body_from_chunks::<std::io::Error>(vec![Ok(Bytes::from(
            "{\"type\":\"ADDED\",\"object\":}\n",
        ))]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        match next_event(&mut stream).await {
            Some(Err(Error::SerdeError(err))) => assert!(!err.is_eof(), "{err}"),
            other => panic!("expected a non-eof serde error, got {other:?}"),
        }
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_nested_non_eof_io_error_is_read_events() {
        let body = body_from_chunks(vec![
            Ok(pod_event_frame("ADDED", "55")),
            Err(nested_io(
                &["outer transport interruption", "inner transport interruption"],
                std::io::Error::new(ErrorKind::ConnectionReset, "socket reset"),
            )),
        ]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "ADDED", "55");
        let io_err = expect_read_events(next_event(&mut stream).await);
        assert_eq!(io_err.kind(), ErrorKind::Other);
        let retained = io_err.get_ref().expect("original body error is the source");
        let body_err = retained.downcast_ref::<Error>().expect("source is the body error");
        assert!(
            chain_has_io_kind(body_err, ErrorKind::ConnectionReset),
            "original source chain dropped ConnectionReset: {body_err:?}"
        );
        assert!(
            chain_has_message(body_err, "outer transport interruption"),
            "{body_err:?}"
        );
        assert!(
            chain_has_message(body_err, "inner transport interruption"),
            "{body_err:?}"
        );
        spawned.await.unwrap();
    }

    #[tokio::test]
    async fn watch_close_notify_message_with_other_io_kind_is_read_events() {
        let body = body_from_chunks(vec![
            Ok(pod_event_frame("MODIFIED", "66")),
            Err(std::io::Error::new(
                ErrorKind::InvalidData,
                "peer closed connection without sending TLS close_notify",
            )),
        ]);
        let (spawned, stream) = open_watch(body).await;
        let mut stream = pin!(stream);

        expect_pod(next_event(&mut stream).await, "MODIFIED", "66");
        let io_err = expect_read_events(next_event(&mut stream).await);
        assert_eq!(io_err.kind(), ErrorKind::Other);
        let retained = io_err.get_ref().expect("original body error is the source");
        let body_err = retained.downcast_ref::<Error>().expect("source is the body error");
        assert!(
            chain_has_message(body_err, "close_notify"),
            "fixture must mention close_notify, got {body_err}"
        );
        assert!(
            chain_has_io_kind(body_err, ErrorKind::InvalidData),
            "original InvalidData kind dropped: {body_err:?}"
        );
        assert!(
            !chain_has_io_kind(body_err, ErrorKind::UnexpectedEof),
            "a close_notify message with a different kind was treated as eof: {body_err:?}"
        );
        spawned.await.unwrap();
    }

    fn pod_event_json(action: &str, resource_version: &str) -> String {
        serde_json::json!({
            "type": action,
            "object": {
                "apiVersion": "v1",
                "kind": "Pod",
                "metadata": {
                    "name": "test",
                    "namespace": "default",
                    "resourceVersion": resource_version,
                },
                "spec": {
                    "containers": [{ "name": "test", "image": "test-image" }],
                }
            }
        })
        .to_string()
    }

    fn pod_event_frame_str(action: &str, resource_version: &str) -> String {
        format!("{}\n", pod_event_json(action, resource_version))
    }

    fn pod_event_frame(action: &str, resource_version: &str) -> Bytes {
        Bytes::from(pod_event_frame_str(action, resource_version))
    }

    fn body_from_chunks<E>(chunks: Vec<Result<Bytes, E>>) -> Body
    where
        E: std::error::Error + Send + Sync + 'static,
    {
        let frames = chunks.into_iter().map(|chunk| chunk.map(Frame::data));
        Body::wrap_body(StreamBody::new(futures::stream::iter(frames)))
    }

    /// `Api::watch` calls `Client::request_events`.
    async fn open_watch(body: Body) -> (tokio::task::JoinHandle<()>, impl Stream<Item = Result<WatchEvent<Pod>>>) {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (request, send) = handle.next_request().await.expect("service not called");
            assert!(
                request.uri().query().unwrap_or_default().contains("watch=true"),
                "expected the client watch path, got {}",
                request.uri()
            );
            send.send_response(Response::new(body));
        });
        let pods: Api<Pod> = Api::default_namespaced(Client::new(mock_service, "default"));
        let stream = pods.watch(&Default::default(), "0").await.expect("watch response");
        (spawned, stream)
    }

    async fn next_event<S>(stream: &mut S) -> Option<Result<WatchEvent<Pod>>>
    where
        S: Stream<Item = Result<WatchEvent<Pod>>> + Unpin,
    {
        tokio::time::timeout(WATCH_POLL_TIMEOUT, stream.next())
            .await
            .expect("watch stream stalled")
    }

    fn expect_pod(item: Option<Result<WatchEvent<Pod>>>, action: &str, resource_version: &str) {
        let pod = match (action, item) {
            ("ADDED", Some(Ok(WatchEvent::Added(pod)))) => pod,
            ("MODIFIED", Some(Ok(WatchEvent::Modified(pod)))) => pod,
            (_, other) => panic!("expected {action} pod, got {other:?}"),
        };
        assert_eq!(pod.metadata.name.as_deref(), Some("test"));
        assert_eq!(
            pod.metadata.resource_version.as_deref(),
            Some(resource_version),
            "completed event must keep its resource version"
        );
    }

    fn expect_end(item: Option<Result<WatchEvent<Pod>>>) {
        assert!(item.is_none(), "expected the watch stream to end, got {item:?}");
    }

    fn expect_read_events(item: Option<Result<WatchEvent<Pod>>>) -> std::io::Error {
        match item {
            Some(Err(Error::ReadEvents(err))) => err,
            other => panic!("expected ReadEvents, got {other:?}"),
        }
    }

    fn nested_io(messages: &[&'static str], io_err: std::io::Error) -> SourceChain {
        let (outer, rest) = messages.split_first().expect("at least one wrapper");
        let mut source: Box<dyn std::error::Error + Send + Sync> = Box::new(io_err);
        for message in rest.iter().rev() {
            source = Box::new(SourceChain {
                message,
                source: Some(source),
            });
        }
        SourceChain {
            message: outer,
            source: Some(source),
        }
    }

    fn chain_has_io_kind(err: &(dyn std::error::Error + 'static), kind: ErrorKind) -> bool {
        let mut current = Some(err);
        for _ in 0..16 {
            let Some(err) = current else {
                return false;
            };
            if let Some(io_err) = err.downcast_ref::<std::io::Error>()
                && io_err.kind() == kind
            {
                return true;
            }
            current = err.source();
        }
        false
    }

    fn chain_has_message(err: &(dyn std::error::Error + 'static), message: &str) -> bool {
        let mut current = Some(err);
        for _ in 0..16 {
            let Some(err) = current else {
                return false;
            };
            if err.to_string().contains(message) {
                return true;
            }
            current = err.source();
        }
        false
    }

    /// Models a `source()` chain without matching on error text.
    struct SourceChain {
        message: &'static str,
        source: Option<Box<dyn std::error::Error + Send + Sync>>,
    }

    impl fmt::Debug for SourceChain {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.debug_struct("SourceChain")
                .field("message", &self.message)
                .field("source", &self.source.as_ref().map(ToString::to_string))
                .finish()
        }
    }

    impl fmt::Display for SourceChain {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str(self.message)
        }
    }

    impl std::error::Error for SourceChain {
        fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
            self.source.as_ref().map(|err| err.as_ref() as _)
        }
    }

    /// Legacy chunk-decoder failure. Its display matches the existing text
    /// check and it is not an `UnexpectedEof` I/O error.
    #[derive(Debug)]
    struct ChunkDecoderEof;

    impl fmt::Display for ChunkDecoderEof {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str("unexpected EOF during chunk size line")
        }
    }

    impl std::error::Error for ChunkDecoderEof {}

    #[cfg(feature = "ws")]
    #[tokio::test]
    async fn connect_returns_api_error_for_forbidden_status() {
        let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
        let spawned = tokio::spawn(async move {
            let mut handle = pin!(handle);
            let (_request, send) = handle.next_request().await.expect("service not called");
            let body = serde_json::json!({
                "status": "Failure",
                "message": "pods \"my-pod\" is forbidden: User \"system:serviceaccount:default:my-sa\" cannot create resource \"pods/exec\"",
                "reason": "Forbidden",
                "code": 403
            });
            send.send_response(
                Response::builder()
                    .status(403)
                    .body(Body::from(serde_json::to_vec(&body).unwrap()))
                    .unwrap(),
            );
        });

        let client = Client::new(mock_service, "default");
        let request = Request::builder()
            .uri("/api/v1/namespaces/default/pods/my-pod/exec")
            .body(Vec::new())
            .unwrap();
        let err = match client.connect(request).await {
            Err(err) => err,
            Ok(_) => panic!("403 is an API error"),
        };
        match err {
            Error::Api(status) => {
                assert_eq!(status.code, 403);
                assert!(status.message.contains("forbidden"), "{}", status.message);
            }
            other => panic!("expected Api error, got {other}"),
        }
        spawned.await.unwrap();
    }
}