xianyun_backends/kubectl/

mod.rs

//! kubectl operations.
//!
//! Ported from `internal/backends/kubectl/{interface,client}.go`.

use std::sync::Arc;
use std::time::Duration;

use serde::Deserialize;
use thiserror::Error;
use xianyun_executor::exec::{CommandRunner, ExecError};

/// Ported from `VersionResponse` in `interface.go`. Every field is
/// `#[serde(default)]`, matching Go's `encoding/json`, which leaves
/// unset fields as their zero value rather than erroring -- kubectl's
/// actual `version -o json` output includes every field, but the
/// ported tests only stub the subset each case cares about.
#[derive(Debug, Default, Deserialize)]
pub struct VersionResponse {
    #[serde(rename = "clientVersion", default)]
    pub client_version: ClientVersion,
    #[serde(rename = "kustomizeVersion", default)]
    pub kustomize_version: String,
    #[serde(rename = "serverVersion", default)]
    pub server_version: ServerVersion,
}

/// Ported from `ClientVersion` in `interface.go`.
#[derive(Debug, Default, Deserialize)]
pub struct ClientVersion {
    #[serde(default)]
    pub major: String,
    #[serde(default)]
    pub minor: String,
    #[serde(rename = "gitVersion", default)]
    pub git_version: String,
    #[serde(rename = "gitCommit", default)]
    pub git_commit: String,
    #[serde(rename = "gitTreeState", default)]
    pub git_tree_state: String,
    #[serde(rename = "goVersion", default)]
    pub go_version: String,
    #[serde(default)]
    pub compiler: String,
    #[serde(default)]
    pub platform: String,
}

/// Ported from `ServerVersion` in `interface.go`.
#[derive(Debug, Default, Deserialize)]
pub struct ServerVersion {
    #[serde(default)]
    pub major: String,
    #[serde(default)]
    pub minor: String,
    #[serde(rename = "gitVersion", default)]
    pub git_version: String,
    #[serde(rename = "gitCommit", default)]
    pub git_commit: String,
    #[serde(rename = "gitTreeState", default)]
    pub git_tree_state: String,
    #[serde(rename = "goVersion", default)]
    pub go_version: String,
    #[serde(default)]
    pub compiler: String,
    #[serde(default)]
    pub platform: String,
}

/// Ported from `ErrNamespaceRequired`.
#[derive(Debug, Error)]
pub enum KubectlError {
    #[error("namespace is required")]
    NamespaceRequired,

    /// Ported from `ErrJobFailed`.
    #[error("job reached the Failed condition: {0}")]
    JobFailed(String),

    #[error(transparent)]
    Exec(#[from] ExecError),

    #[error("{0}")]
    Other(String),
}

/// Ported from the `Backend` interface in `interface.go`.
pub trait Backend {
    fn get(&self, resource_type: &str, name: &str, namespace: &str)
    -> Result<String, KubectlError>;

    fn delete_pods_by_label(
        &self,
        namespace: &str,
        label_selector: &str,
    ) -> Result<usize, KubectlError>;

    fn get_version(
        &self,
        timeout: Duration,
        interval: Duration,
    ) -> Result<VersionResponse, KubectlError>;

    fn scale(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
        replicas: i64,
    ) -> Result<(), KubectlError>;

    fn apply(&self, json_data: &[u8], namespace: &str) -> Result<(), KubectlError>;

    fn delete_manifest(&self, data: &[u8], namespace: &str) -> Result<(), KubectlError>;

    fn delete(&self, resource_type: &str, name: &str, namespace: &str) -> Result<(), KubectlError>;

    fn wait(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
        condition: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError>;

    fn wait_for_rollout_status(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError>;

    fn wait_for_pod_ready_by_label(
        &self,
        namespace: &str,
        label_selector: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError>;

    fn wait_for_job_complete(
        &self,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError>;

    fn wait_for_job_created(
        &self,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError>;

    fn list_resource_names(
        &self,
        kind: &str,
        namespace: &str,
        label_selector: &str,
    ) -> Result<Vec<String>, KubectlError>;

    fn ensure_namespace(&self, name: &str) -> Result<(), KubectlError>;

    fn delete_namespace(&self, name: &str) -> Result<(), KubectlError>;

    fn namespace_exists(&self, name: &str) -> Result<bool, KubectlError>;

    fn wait_for_namespace_ready(&self, name: &str, timeout: Duration) -> Result<(), KubectlError>;

    fn run_job(
        &self,
        image: &str,
        namespace: &str,
        timeout: Duration,
        command: &[&str],
    ) -> Result<String, KubectlError>;

    fn get_raw(&self, path: &str) -> Result<String, KubectlError>;
}

/// Ported from `Client` in `client.go`.
pub struct Client {
    kubeconfig: Option<String>,
    executor: Arc<dyn CommandRunner + Send + Sync>,
}

/// A kubeconfig materialized to a throwaway temp file so it can be
/// pointed at via a `KUBECONFIG` env var -- the executed `kubectl`
/// process reads the file path, not the in-memory string. Deletes the
/// file on drop. Mirrors `k3d::Driver`'s `TempKubeconfigGuard` (see its
/// doc comment in `crates/xianyun-cluster/src/driver/k3d.rs`), which
/// predates this one and does the same thing for the same reason.
struct KubeconfigFile(std::path::PathBuf);

impl Drop for KubeconfigFile {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.0);
    }
}

impl Client {
    /// Ported from `NewClient`. Go's zero-argument `NewClient` reads the
    /// command runner from the ambient `context.Context` on every call
    /// instead of taking it at construction; this port takes it
    /// explicitly, matching `docker::Client`/`git::Client`. See
    /// `PORT_NOTES.md`.
    pub fn new(executor: Arc<dyn CommandRunner + Send + Sync>) -> Self {
        Self {
            kubeconfig: None,
            executor,
        }
    }

    /// Ported from `NewClientWithConfig`.
    pub fn with_config(executor: Arc<dyn CommandRunner + Send + Sync>, kubeconfig: &str) -> Self {
        Self {
            kubeconfig: Some(kubeconfig.to_string()),
            executor,
        }
    }

    /// Ported from `Client.kubeconfigEnv`: when this client carries a
    /// kubeconfig (`with_config`, not `new`), writes it to a throwaway
    /// temp file and returns a `KUBECONFIG=<path>` env var appended onto
    /// the current process's own environment (so `kubectl` still finds
    /// `PATH` and everything else it needs), plus a guard that deletes
    /// the file once the caller is done with it. Returns `None` when
    /// this client has no kubeconfig (`new`'s case): every caller below
    /// falls back to `kubectl`'s own default resolution (ambient
    /// `KUBECONFIG`/`~/.kube/config`) exactly as before, unchanged.
    fn kubeconfig_env(&self) -> Result<Option<(Vec<String>, KubeconfigFile)>, KubectlError> {
        let Some(kubeconfig) = &self.kubeconfig else {
            return Ok(None);
        };

        let pid = std::process::id();
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or_default();
        let path = std::env::temp_dir().join(format!("xianyun-kubeconfig-{pid}-{nanos}"));

        std::fs::write(&path, kubeconfig)
            .map_err(|err| KubectlError::Other(format!("writing temp kubeconfig: {err}")))?;

        let mut env: Vec<String> = std::env::vars().map(|(k, v)| format!("{k}={v}")).collect();
        env.push(format!("KUBECONFIG={}", path.display()));

        Ok(Some((env, KubeconfigFile(path))))
    }

    /// Ported from `Client.readStdout`. Materializes `self.kubeconfig`
    /// (when set) into a `KUBECONFIG=<path>` env var via
    /// [`Self::kubeconfig_env`] before running, matching Go's
    /// `kubeconfigEnv` call in the original `readStdout`.
    fn read_stdout(&self, name: &str, args: &[String]) -> Result<String, KubectlError> {
        let env_guard = self.kubeconfig_env()?;
        let opts = env_guard
            .as_ref()
            .map(|(env, _)| xianyun_executor::exec::Options {
                env: Some(env.clone()),
                ..Default::default()
            });

        let result = self.executor.run(name, args, opts);
        if result.error.is_some() {
            return Err(result_error(result));
        }

        Ok(result.stdout.join("\n"))
    }

    /// Ported from `Client.runManifestCommand`. Runs a kubectl verb that
    /// reads a manifest bundle from stdin (`apply`, `delete`, ...). An
    /// empty namespace lets each object use its own
    /// `metadata.namespace` (and leaves cluster-scoped objects alone).
    fn run_manifest_command(
        &self,
        verb: &str,
        data: &[u8],
        namespace: &str,
        extra_args: &[&str],
    ) -> Result<(), KubectlError> {
        let mut args = vec![verb.to_string()];
        if !namespace.is_empty() {
            args.push("-n".to_string());
            args.push(namespace.to_string());
        }
        args.push("-f".to_string());
        args.push("-".to_string());
        args.extend(extra_args.iter().map(std::string::ToString::to_string));

        let env_guard = self.kubeconfig_env()?;

        let opts = xianyun_executor::exec::Options {
            stdin: Some(Box::new(std::io::Cursor::new(data.to_vec()))),
            env: env_guard.as_ref().map(|(env, _)| env.clone()),
            ..Default::default()
        };

        let result = self.executor.run("kubectl", &args, Some(opts));
        if result.error.is_some() {
            return Err(result_error(result));
        }

        Ok(())
    }
}

/// Ported from `resultError`: wraps `result.error` with the last
/// non-empty stderr line, if any, so callers see kubectl's actual
/// message (e.g. "Error from server (Forbidden): ...") rather than only
/// the exit status. Returns `KubectlError::Exec` untouched when there is
/// no stderr to append, so `matches!(err, KubectlError::Exec(..))`
/// assertions on a bare exec failure keep working; otherwise returns
/// `KubectlError::Other`, since `ExecError` has no variant carrying an
/// extra message.
///
/// # Panics
///
/// Panics if `result.error` is `None`; callers must only invoke this
/// after checking `result.error.is_some()`.
fn result_error(result: xianyun_executor::exec::ExecResult) -> KubectlError {
    let err = result
        .error
        .expect("result_error requires result.error to be Some");

    for line in result.stderr.iter().rev() {
        let trimmed = line.trim();
        if !trimmed.is_empty() {
            return KubectlError::Other(format!("{err}: {trimmed}"));
        }
    }

    KubectlError::Exec(err)
}

/// Ported from `transientAPIErrorMarkers`/`isTransientAPIError`:
/// substrings kubectl writes to stderr when the API server endpoint is
/// not yet reachable, matched case-insensitively.
const TRANSIENT_API_ERROR_MARKERS: &[&str] = &[
    "was refused",
    "connection refused",
    "unable to connect to the server",
    "connection reset",
    "i/o timeout",
    "tls handshake timeout",
    "server is currently unable to handle the request",
];

fn is_transient_api_error(stderr: &[String]) -> bool {
    let haystack = stderr.join("\n").to_lowercase();
    TRANSIENT_API_ERROR_MARKERS
        .iter()
        .any(|marker| haystack.contains(marker))
}

impl Backend for Client {
    fn get(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
    ) -> Result<String, KubectlError> {
        let mut args = vec!["get".to_string(), resource_type.to_string()];
        if !name.is_empty() {
            args.push(name.to_string());
        }
        args.push("-n".to_string());
        args.push(namespace.to_string());
        args.push("-o".to_string());
        args.push("json".to_string());

        self.read_stdout("kubectl", &args)
    }

    fn delete_pods_by_label(
        &self,
        namespace: &str,
        label_selector: &str,
    ) -> Result<usize, KubectlError> {
        let output = self.read_stdout(
            "kubectl",
            &[
                "get".to_string(),
                "pods".to_string(),
                "-n".to_string(),
                namespace.to_string(),
                "-l".to_string(),
                label_selector.to_string(),
                "-o".to_string(),
                "jsonpath={.items[*].metadata.name}".to_string(),
            ],
        )?;

        let names: Vec<&str> = output.split_whitespace().collect();
        if names.is_empty() {
            return Ok(0);
        }

        let mut args = vec![
            "delete".to_string(),
            "pod".to_string(),
            "-n".to_string(),
            namespace.to_string(),
            "--wait=false".to_string(),
        ];
        args.extend(names.iter().map(std::string::ToString::to_string));

        self.read_stdout("kubectl", &args)?;

        Ok(names.len())
    }

    fn get_version(
        &self,
        timeout: Duration,
        interval: Duration,
    ) -> Result<VersionResponse, KubectlError> {
        let args = [
            "version".to_string(),
            "-o".to_string(),
            "json".to_string(),
            "--request-timeout=5s".to_string(),
        ];
        let deadline = std::time::Instant::now() + timeout;

        let env_guard = self.kubeconfig_env()?;
        let env = env_guard.as_ref().map(|(env, _)| env.clone());

        loop {
            let opts = env.as_ref().map(|env| xianyun_executor::exec::Options {
                env: Some(env.clone()),
                ..Default::default()
            });
            let result = self.executor.run("kubectl", &args, opts);

            if result.error.is_none() {
                let joined = result.stdout.join("\n");
                let response: VersionResponse = serde_json::from_str(&joined).map_err(|err| {
                    KubectlError::Other(format!("parsing kubectl version output: {err}"))
                })?;

                return Ok(response);
            }

            if !is_transient_api_error(&result.stderr) {
                return Err(result_error(result));
            }

            if std::time::Instant::now() >= deadline {
                let inner = result_error(result);
                return Err(KubectlError::Other(format!(
                    "cluster API not ready after {timeout:?}: {inner}"
                )));
            }

            std::thread::sleep(interval);
        }
    }

    fn scale(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
        replicas: i64,
    ) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "scale".to_string(),
                "--replicas".to_string(),
                replicas.to_string(),
                resource_type.to_string(),
                name.to_string(),
                "-n".to_string(),
                namespace.to_string(),
            ],
        )?;

        Ok(())
    }

    fn apply(&self, json_data: &[u8], namespace: &str) -> Result<(), KubectlError> {
        self.run_manifest_command("apply", json_data, namespace, &[])
    }

    fn delete_manifest(&self, data: &[u8], namespace: &str) -> Result<(), KubectlError> {
        self.run_manifest_command("delete", data, namespace, &["--ignore-not-found"])
    }

    fn delete(&self, resource_type: &str, name: &str, namespace: &str) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "delete".to_string(),
                resource_type.to_string(),
                name.to_string(),
                "-n".to_string(),
                namespace.to_string(),
            ],
        )?;

        Ok(())
    }

    fn wait(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
        condition: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "wait".to_string(),
                format!("--for=condition={condition}"),
                format!("{resource_type}/{name}"),
                "-n".to_string(),
                namespace.to_string(),
                format!("--timeout={}", format_go_duration(timeout)),
            ],
        )?;

        Ok(())
    }

    fn wait_for_rollout_status(
        &self,
        resource_type: &str,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "rollout".to_string(),
                "status".to_string(),
                format!("{resource_type}/{name}"),
                "-n".to_string(),
                namespace.to_string(),
                format!("--timeout={}", format_go_duration(timeout)),
            ],
        )?;

        Ok(())
    }

    fn wait_for_pod_ready_by_label(
        &self,
        namespace: &str,
        label_selector: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "wait".to_string(),
                "--for=condition=Ready".to_string(),
                "pod".to_string(),
                "-l".to_string(),
                label_selector.to_string(),
                "-n".to_string(),
                namespace.to_string(),
                format!("--timeout={}", format_go_duration(timeout)),
            ],
        )?;

        Ok(())
    }

    fn wait_for_job_complete(
        &self,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "wait".to_string(),
                "--for=condition=Complete".to_string(),
                format!("job/{name}"),
                "-n".to_string(),
                namespace.to_string(),
                format!("--timeout={}", format_go_duration(timeout)),
            ],
        )?;

        Ok(())
    }

    fn wait_for_job_created(
        &self,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "wait".to_string(),
                "--for=create".to_string(),
                format!("job/{name}"),
                "-n".to_string(),
                namespace.to_string(),
                format!("--timeout={}", format_go_duration(timeout)),
            ],
        )?;

        Ok(())
    }

    fn list_resource_names(
        &self,
        kind: &str,
        namespace: &str,
        label_selector: &str,
    ) -> Result<Vec<String>, KubectlError> {
        if namespace.is_empty() {
            return Err(KubectlError::NamespaceRequired);
        }

        let output = self.read_stdout(
            "kubectl",
            &[
                "get".to_string(),
                kind.to_string(),
                "-n".to_string(),
                namespace.to_string(),
                "-l".to_string(),
                label_selector.to_string(),
                "-o".to_string(),
                "jsonpath={.items[*].metadata.name}".to_string(),
            ],
        )?;

        Ok(output.split_whitespace().map(str::to_string).collect())
    }

    fn ensure_namespace(&self, name: &str) -> Result<(), KubectlError> {
        let manifest =
            format!(r#"{{"apiVersion":"v1","kind":"Namespace","metadata":{{"name":"{name}"}}}}"#);

        self.run_manifest_command("apply", manifest.as_bytes(), "", &[])
            .map_err(|err| KubectlError::Other(format!("ensuring namespace {name:?}: {err}")))
    }

    fn delete_namespace(&self, name: &str) -> Result<(), KubectlError> {
        self.read_stdout(
            "kubectl",
            &[
                "delete".to_string(),
                "namespace".to_string(),
                name.to_string(),
                "--ignore-not-found".to_string(),
            ],
        )
        .map(|_| ())
        .map_err(|err| KubectlError::Other(format!("deleting namespace {name:?}: {err}")))
    }

    fn namespace_exists(&self, name: &str) -> Result<bool, KubectlError> {
        let output = self
            .read_stdout(
                "kubectl",
                &[
                    "get".to_string(),
                    "namespace".to_string(),
                    name.to_string(),
                    "--ignore-not-found".to_string(),
                    "-o".to_string(),
                    "name".to_string(),
                ],
            )
            .map_err(|err| KubectlError::Other(format!("checking namespace {name:?}: {err}")))?;

        Ok(!output.trim().is_empty())
    }

    fn wait_for_namespace_ready(&self, name: &str, timeout: Duration) -> Result<(), KubectlError> {
        // Ported from `Client.WaitForNamespaceReady`: both `kubectl
        // wait` calls share a single deadline, so the second call's
        // `--timeout=` is whatever remains of `timeout` after the
        // first call returns, floored at zero. `format_go_duration`
        // rounds up to whole seconds, so with negligible elapsed time
        // (as in every ported unit test) the first call's timeout still
        // renders as the exact original value (e.g. "30s").
        let deadline = std::time::Instant::now() + timeout;
        let remaining_timeout_arg = || {
            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
            format!("--timeout={}", format_go_duration(remaining))
        };

        self.read_stdout(
            "kubectl",
            &[
                "wait".to_string(),
                "--for=jsonpath={.status.phase}=Active".to_string(),
                format!("namespace/{name}"),
                remaining_timeout_arg(),
            ],
        )
        .map_err(|err| {
            KubectlError::Other(format!(
                "waiting for namespace {name:?} to be Active: {err}"
            ))
        })?;

        self.read_stdout(
            "kubectl",
            &[
                "wait".to_string(),
                "--for=create".to_string(),
                "serviceaccount/default".to_string(),
                "-n".to_string(),
                name.to_string(),
                remaining_timeout_arg(),
            ],
        )
        .map_err(|err| {
            KubectlError::Other(format!(
                "waiting for default ServiceAccount in namespace {name:?}: {err}"
            ))
        })?;

        Ok(())
    }

    fn run_job(
        &self,
        image: &str,
        namespace: &str,
        timeout: Duration,
        command: &[&str],
    ) -> Result<String, KubectlError> {
        let name = format!(
            "caproni-probe-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos()
        );

        #[allow(clippy::cast_possible_truncation)]
        let active_deadline_seconds = timeout.as_secs_f64().ceil().max(1.0) as i64;
        let command_json = serde_json::to_string(command)
            .map_err(|err| KubectlError::Other(format!("encoding job {name:?}: {err}")))?;
        let manifest = format!(
            r#"{{"apiVersion":"batch/v1","kind":"Job","metadata":{{"name":"{name}","namespace":"{namespace}"}},"spec":{{"backoffLimit":0,"ttlSecondsAfterFinished":60,"activeDeadlineSeconds":{active_deadline_seconds},"template":{{"spec":{{"restartPolicy":"Never","containers":[{{"name":"probe","image":"{image}","command":{command_json}}}]}}}}}}}}"#,
        );

        self.run_manifest_command("create", manifest.as_bytes(), namespace, &[])
            .map_err(|err| KubectlError::Other(format!("creating job {name:?}: {err}")))?;

        let cleanup = || {
            let _ = self.read_stdout(
                "kubectl",
                &[
                    "delete".to_string(),
                    "job".to_string(),
                    name.clone(),
                    "-n".to_string(),
                    namespace.to_string(),
                    "--ignore-not-found".to_string(),
                ],
            );
        };

        let wait_result = self.wait_for_job_finished(&name, namespace, timeout);
        if let Err(err) = wait_result {
            cleanup();
            return Err(KubectlError::Other(format!("running job {name:?}: {err}")));
        }

        let output = self
            .read_stdout(
                "kubectl",
                &[
                    "logs".to_string(),
                    format!("job/{name}"),
                    "-n".to_string(),
                    namespace.to_string(),
                ],
            )
            .map_err(|err| KubectlError::Other(format!("reading logs of job {name:?}: {err}")));

        cleanup();

        output
    }

    fn get_raw(&self, path: &str) -> Result<String, KubectlError> {
        self.read_stdout(
            "kubectl",
            &["get".to_string(), "--raw".to_string(), path.to_string()],
        )
        .map_err(|err| KubectlError::Other(format!("getting raw path {path:?}: {err}")))
    }
}

/// Ported from Go's `time.Duration.String()`, the format
/// `--timeout=<duration>` needs (`kubectl wait --help` accepts the same
/// syntax as Go's `time.ParseDuration`). This only needs to round-trip
/// through kubectl, not match Go's formatting rules exactly (which
/// picks the coarsest unit and can combine units, e.g. "1m30s"); a
/// plain `<seconds>s` is unambiguous input.
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn format_go_duration(d: Duration) -> String {
    // Rounds up to whole seconds so a duration that is nominally "30s"
    // but has had a handful of microseconds shaved off it by real
    // wall-clock elapsed time (see `wait_for_namespace_ready`'s shared
    // deadline) still formats as "30s" rather than "29.999998s".
    format!("{}s", d.as_secs_f64().ceil() as u64)
}

impl Client {
    /// Ported from `Client.waitForJobFinished`. Polls the Job's status
    /// until it reaches `Complete` (returns `Ok`), `Failed` (returns
    /// `KubectlError::JobFailed`), or `timeout` elapses.
    fn wait_for_job_finished(
        &self,
        name: &str,
        namespace: &str,
        timeout: Duration,
    ) -> Result<(), KubectlError> {
        let deadline = std::time::Instant::now() + timeout;

        loop {
            let out = self.get("job", name, namespace)?;
            let status: JobStatus = serde_json::from_str(&out)
                .map_err(|err| KubectlError::Other(format!("parsing job status: {err}")))?;

            for cond in &status.status.conditions {
                if cond.status != "True" {
                    continue;
                }

                match cond.type_.as_str() {
                    "Complete" => return Ok(()),
                    "Failed" => {
                        return Err(KubectlError::JobFailed(format!(
                            "{}: {}",
                            cond.reason, cond.message
                        )));
                    }
                    _ => {}
                }
            }

            if std::time::Instant::now() >= deadline {
                return Err(KubectlError::Other(format!(
                    "waiting for job {name} to finish: deadline exceeded"
                )));
            }

            std::thread::sleep(Duration::from_millis(50));
        }
    }
}

/// Ported from `jobStatus`: the part of a Job's status that says
/// whether it has finished.
#[derive(Debug, Deserialize, Default)]
struct JobStatus {
    #[serde(default)]
    status: JobStatusInner,
}

#[derive(Debug, Deserialize, Default)]
struct JobStatusInner {
    #[serde(default)]
    conditions: Vec<JobCondition>,
}

#[derive(Debug, Deserialize)]
struct JobCondition {
    #[serde(rename = "type")]
    type_: String,
    status: String,
    #[serde(default)]
    reason: String,
    #[serde(default)]
    message: String,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Read as _;
    use std::path::PathBuf;
    use std::sync::Mutex;

    use xianyun_executor::exec::{ExecResult, ExecStreamedOpts};
    use xianyun_testutil::MockCommandRunner;

    fn new_client(mock: MockCommandRunner) -> Client {
        Client::new(Arc::new(mock))
    }

    /// A local, single-purpose `CommandRunner` that captures whatever
    /// `Options::env` it was called with, for the `*_sets_kubeconfig_env`
    /// tests below -- `xianyun_testutil::MockCommandRunner::run` ignores
    /// `opts` entirely (see its doc comment), so a `with_config` client's
    /// one new behavior (pointing `KUBECONFIG` at a temp file) needs its
    /// own capturing runner, the same way `StdinCapturingRunner` exists
    /// for the handful of tests needing `Options::stdin` instead.
    struct EnvCapturingRunner {
        result: Mutex<Option<ExecResult>>,
        captured_env: Mutex<Option<Vec<String>>>,
    }

    impl EnvCapturingRunner {
        fn new(result: ExecResult) -> Self {
            Self {
                result: Mutex::new(Some(result)),
                captured_env: Mutex::new(None),
            }
        }
    }

    impl CommandRunner for EnvCapturingRunner {
        fn run(
            &self,
            _name: &str,
            _args: &[String],
            opts: Option<xianyun_executor::exec::Options>,
        ) -> ExecResult {
            *self.captured_env.lock().unwrap() = opts.and_then(|o| o.env);
            self.result
                .lock()
                .unwrap()
                .take()
                .expect("EnvCapturingRunner::run invoked more than once")
        }

        fn read_stdout(&self, _name: &str, _args: &[String]) -> Result<String, ExecError> {
            unimplemented!("EnvCapturingRunner only supports run()")
        }

        fn run_sudo(&self, _name: &str, _args: &[String]) -> Result<(), ExecError> {
            unimplemented!("EnvCapturingRunner only supports run()")
        }

        fn check(&self, _name: &str, _args: &[String]) -> Result<bool, ExecError> {
            unimplemented!("EnvCapturingRunner only supports run()")
        }

        fn exec_streamed_command(
            &self,
            _opts: Option<ExecStreamedOpts>,
            _name: &str,
            _args: &[String],
        ) -> Result<(), ExecError> {
            unimplemented!("EnvCapturingRunner only supports run()")
        }

        fn look_path(&self, _file: &str) -> Result<PathBuf, ExecError> {
            unimplemented!("EnvCapturingRunner only supports run()")
        }

        fn run_in_repository(
            &self,
            _repo_dir: &str,
            _name: &str,
            _args: &[String],
            _opts: Option<ExecStreamedOpts>,
        ) -> Result<(), ExecError> {
            unimplemented!("EnvCapturingRunner only supports run()")
        }
    }

    // --- New coverage: KUBECONFIG env wiring for `Client::with_config` ---
    //
    // Closes the gap `kubeconfig_env`'s doc comment (and previously this
    // module's `Client` doc comment) described: a `with_config` client
    // materializes its kubeconfig into a temp file and points
    // `KUBECONFIG` at it on every command, instead of silently falling
    // back to kubectl's own default resolution. Confirmed as a real,
    // user-visible bug (not just a documentation gap) by the first real
    // run of `nix/tests/e2e-k3d.nix` (see book/src/project/testing.md): `xianyun
    // up` against a real k3d cluster failed with "the connection to the
    // server localhost:8080 was refused" -- `kubectl`'s hardcoded
    // fallback when it finds no kubeconfig anywhere, which it hit
    // because nothing ever told it which cluster's kubeconfig to use.

    #[test]
    fn read_stdout_path_sets_kubeconfig_env_when_with_config() {
        let runner = Arc::new(EnvCapturingRunner::new(ExecResult {
            stdout: vec!["ns-a".to_string()],
            ..Default::default()
        }));
        let client = Client::with_config(runner.clone(), "apiVersion: v1\nkind: Config\n");

        let names = client
            .list_resource_names("pod", "default", "")
            .expect("list_resource_names should succeed");
        assert_eq!(names, vec!["ns-a".to_string()]);

        let env = runner
            .captured_env
            .lock()
            .unwrap()
            .clone()
            .expect("with_config client must set Options::env");
        assert!(
            env.iter().any(|e| e.starts_with("KUBECONFIG=")),
            "expected a KUBECONFIG=... entry, got: {env:?}"
        );
    }

    #[test]
    fn read_stdout_path_sets_no_env_override_when_plain_new() {
        let runner = Arc::new(EnvCapturingRunner::new(ExecResult {
            stdout: vec!["ns-a".to_string()],
            ..Default::default()
        }));
        let client = Client::new(runner.clone());

        client
            .list_resource_names("pod", "default", "")
            .expect("list_resource_names should succeed");

        assert_eq!(
            runner.captured_env.lock().unwrap().clone(),
            None,
            "a plain `new` client (no kubeconfig) must not override the environment"
        );
    }

    #[test]
    fn run_manifest_command_path_sets_kubeconfig_env_when_with_config() {
        let runner = Arc::new(EnvCapturingRunner::new(ExecResult::default()));
        let client = Client::with_config(runner.clone(), "apiVersion: v1\nkind: Config\n");

        client
            .apply(b"kind: ConfigMap\n", "default")
            .expect("apply should succeed");

        let env = runner
            .captured_env
            .lock()
            .unwrap()
            .clone()
            .expect("with_config client must set Options::env");
        assert!(
            env.iter().any(|e| e.starts_with("KUBECONFIG=")),
            "expected a KUBECONFIG=... entry, got: {env:?}"
        );
    }

    #[test]
    fn get_version_path_sets_kubeconfig_env_when_with_config() {
        let runner = Arc::new(EnvCapturingRunner::new(ExecResult {
            stdout: vec![r#"{"serverVersion":{"gitVersion":"v1.32.5+k3s1"}}"#.to_string()],
            ..Default::default()
        }));
        let client = Client::with_config(runner.clone(), "apiVersion: v1\nkind: Config\n");

        client
            .get_version(Duration::from_secs(1), Duration::from_millis(10))
            .expect("get_version should succeed");

        let env = runner
            .captured_env
            .lock()
            .unwrap()
            .clone()
            .expect("with_config client must set Options::env");
        assert!(
            env.iter().any(|e| e.starts_with("KUBECONFIG=")),
            "expected a KUBECONFIG=... entry, got: {env:?}"
        );
    }

    /// A local, single-purpose `CommandRunner` for the handful of tests
    /// that need to inspect what was piped to a call's stdin --
    /// `xianyun_testutil::MockCommandRunner`'s `expect_run*` family
    /// matches only `&[String]` args, with no `Options`/`Stdin`-aware
    /// variant (see `PORT_NOTES.md`'s "Mocking strategy" section), and
    /// `xianyun-testutil` is out of scope for this crate to extend.
    /// Rather than block on that, this reads the piped bytes directly
    /// off `Options::stdin` and asserts on them, exactly the assertion
    /// Go's `mock.MatchedBy(func(opts *executor.Options) bool { ... })`
    /// makes in `TestEnsureNamespace`.
    ///
    /// Every other `CommandRunner` method panics with `unimplemented!()`
    /// since `Client::ensure_namespace` only ever calls `run`.
    struct StdinCapturingRunner {
        expected_name: String,
        expected_args: Vec<String>,
        expected_stdin: String,
        result: Mutex<Option<ExecResult>>,
    }

    impl StdinCapturingRunner {
        fn new(name: &str, args: &[&str], expected_stdin: &str, result: ExecResult) -> Self {
            Self {
                expected_name: name.to_string(),
                expected_args: owned_strings(args),
                expected_stdin: expected_stdin.to_string(),
                result: Mutex::new(Some(result)),
            }
        }
    }

    fn owned_strings(args: &[&str]) -> Vec<String> {
        args.iter().map(|s| (*s).to_string()).collect()
    }

    impl CommandRunner for StdinCapturingRunner {
        fn run(
            &self,
            name: &str,
            args: &[String],
            opts: Option<xianyun_executor::exec::Options>,
        ) -> ExecResult {
            assert_eq!(name, self.expected_name);
            assert_eq!(args, self.expected_args.as_slice());

            let mut opts = opts.expect("caller must pipe the manifest through Options::stdin");
            let mut stdin = opts
                .stdin
                .take()
                .expect("caller must set Options::stdin to the manifest bytes");
            let mut buf = String::new();
            stdin
                .read_to_string(&mut buf)
                .expect("piped stdin must be valid UTF-8");
            assert_eq!(buf, self.expected_stdin);

            self.result
                .lock()
                .unwrap()
                .take()
                .expect("StdinCapturingRunner::run invoked more than once")
        }

        fn read_stdout(&self, _name: &str, _args: &[String]) -> Result<String, ExecError> {
            unimplemented!("StdinCapturingRunner only supports run()")
        }

        fn run_sudo(&self, _name: &str, _args: &[String]) -> Result<(), ExecError> {
            unimplemented!("StdinCapturingRunner only supports run()")
        }

        fn check(&self, _name: &str, _args: &[String]) -> Result<bool, ExecError> {
            unimplemented!("StdinCapturingRunner only supports run()")
        }

        fn exec_streamed_command(
            &self,
            _opts: Option<ExecStreamedOpts>,
            _name: &str,
            _args: &[String],
        ) -> Result<(), ExecError> {
            unimplemented!("StdinCapturingRunner only supports run()")
        }

        fn look_path(&self, _file: &str) -> Result<PathBuf, ExecError> {
            unimplemented!("StdinCapturingRunner only supports run()")
        }

        fn run_in_repository(
            &self,
            _repo_dir: &str,
            _name: &str,
            _args: &[String],
            _opts: Option<ExecStreamedOpts>,
        ) -> Result<(), ExecError> {
            unimplemented!("StdinCapturingRunner only supports run()")
        }
    }

    /// A local `CommandRunner` for the three `RunJob` lifecycle tests.
    /// It captures the generated Job name off the `create` call's piped
    /// manifest (the same "read stdin, then use the decoded name to
    /// shape later expectations" pattern Go's `expectJobCreate` +
    /// `mock.RunAndReturn` implements) and then serves each subsequent
    /// `get`/`logs`/`delete` call by matching on that captured name.
    /// Not a general-purpose mock: it hard-codes the exact call
    /// sequence `Client::run_job` makes, which is exactly what each
    /// ported test needs to assert on.
    struct JobLifecycleRunner {
        namespace: String,
        job_name: Mutex<Option<String>>,
        get_job_responses: Mutex<Vec<String>>,
        get_job_repeat_last: bool,
        logs_stdout: Option<Vec<String>>,
        delete_calls: Mutex<u32>,
    }

    impl JobLifecycleRunner {
        fn new(
            namespace: &str,
            get_job_responses: Vec<&str>,
            logs_stdout: Option<Vec<&str>>,
        ) -> Self {
            Self {
                namespace: namespace.to_string(),
                job_name: Mutex::new(None),
                get_job_responses: Mutex::new(
                    get_job_responses.into_iter().map(str::to_string).collect(),
                ),
                get_job_repeat_last: false,
                logs_stdout: logs_stdout
                    .map(|lines| lines.into_iter().map(str::to_string).collect()),
                delete_calls: Mutex::new(0),
            }
        }

        /// The `run_job_deletes_job_when_deadline_elapses` test polls
        /// `get job` an indeterminate number of times before the
        /// deadline elapses (Go's `.Maybe()`) -- every call gets the
        /// same never-terminal response rather than a fixed sequence.
        fn repeating(namespace: &str, get_job_response: &str) -> Self {
            Self {
                namespace: namespace.to_string(),
                job_name: Mutex::new(None),
                get_job_responses: Mutex::new(vec![get_job_response.to_string()]),
                get_job_repeat_last: true,
                logs_stdout: None,
                delete_calls: Mutex::new(0),
            }
        }

        fn job_name(&self) -> String {
            self.job_name
                .lock()
                .unwrap()
                .clone()
                .expect("job must have been created before this assertion")
        }

        fn delete_call_count(&self) -> u32 {
            *self.delete_calls.lock().unwrap()
        }
    }

    impl CommandRunner for JobLifecycleRunner {
        fn run(
            &self,
            name: &str,
            args: &[String],
            opts: Option<xianyun_executor::exec::Options>,
        ) -> ExecResult {
            assert_eq!(name, "kubectl");
            let verb = args.first().map(String::as_str);

            match verb {
                Some("create") => {
                    assert_eq!(
                        args,
                        ["create", "-n", self.namespace.as_str(), "-f", "-"].as_slice()
                    );

                    let mut opts =
                        opts.expect("RunJob must pipe the Job manifest through Options::stdin");
                    let mut stdin = opts
                        .stdin
                        .take()
                        .expect("RunJob must set Options::stdin to the Job manifest");
                    let mut data = Vec::new();
                    stdin
                        .read_to_end(&mut data)
                        .expect("piped Job manifest must be readable");

                    let manifest: serde_json::Value =
                        serde_json::from_slice(&data).expect("piped Job manifest must be JSON");
                    let job_name = manifest["metadata"]["name"]
                        .as_str()
                        .expect("Job manifest must set metadata.name")
                        .to_string();

                    assert_eq!(manifest["apiVersion"], "batch/v1");
                    assert_eq!(manifest["kind"], "Job");
                    assert_eq!(manifest["metadata"]["namespace"], self.namespace);
                    assert_eq!(manifest["spec"]["backoffLimit"], 0);
                    assert!(
                        manifest["spec"]["ttlSecondsAfterFinished"]
                            .as_i64()
                            .unwrap()
                            > 0
                    );
                    assert_eq!(
                        manifest["spec"]["template"]["spec"]["restartPolicy"],
                        "Never"
                    );

                    *self.job_name.lock().unwrap() = Some(job_name);
                    ExecResult::default()
                }
                Some("get") => {
                    let job_name = self.job_name();
                    assert_eq!(
                        args,
                        [
                            "get",
                            "job",
                            job_name.as_str(),
                            "-n",
                            self.namespace.as_str(),
                            "-o",
                            "json",
                        ]
                        .as_slice()
                    );

                    let mut queue = self.get_job_responses.lock().unwrap();
                    let json = if self.get_job_repeat_last {
                        queue[0].clone()
                    } else {
                        assert!(!queue.is_empty(), "unexpected extra `kubectl get job` call");
                        queue.remove(0)
                    };
                    drop(queue);

                    ExecResult {
                        stdout: vec![json],
                        ..Default::default()
                    }
                }
                Some("logs") => {
                    let job_name = self.job_name();
                    assert_eq!(
                        args,
                        [
                            "logs",
                            format!("job/{job_name}").as_str(),
                            "-n",
                            self.namespace.as_str(),
                        ]
                        .as_slice()
                    );

                    ExecResult {
                        stdout: self
                            .logs_stdout
                            .clone()
                            .expect("this scenario does not expect a `kubectl logs` call"),
                        ..Default::default()
                    }
                }
                Some("delete") => {
                    let job_name = self.job_name();
                    assert_eq!(
                        args,
                        [
                            "delete",
                            "job",
                            job_name.as_str(),
                            "-n",
                            self.namespace.as_str(),
                            "--ignore-not-found",
                        ]
                        .as_slice()
                    );

                    *self.delete_calls.lock().unwrap() += 1;
                    ExecResult::default()
                }
                other => panic!("JobLifecycleRunner: unexpected kubectl verb {other:?}"),
            }
        }

        fn read_stdout(&self, _name: &str, _args: &[String]) -> Result<String, ExecError> {
            unimplemented!("JobLifecycleRunner only supports run()")
        }

        fn run_sudo(&self, _name: &str, _args: &[String]) -> Result<(), ExecError> {
            unimplemented!("JobLifecycleRunner only supports run()")
        }

        fn check(&self, _name: &str, _args: &[String]) -> Result<bool, ExecError> {
            unimplemented!("JobLifecycleRunner only supports run()")
        }

        fn exec_streamed_command(
            &self,
            _opts: Option<ExecStreamedOpts>,
            _name: &str,
            _args: &[String],
        ) -> Result<(), ExecError> {
            unimplemented!("JobLifecycleRunner only supports run()")
        }

        fn look_path(&self, _file: &str) -> Result<PathBuf, ExecError> {
            unimplemented!("JobLifecycleRunner only supports run()")
        }

        fn run_in_repository(
            &self,
            _repo_dir: &str,
            _name: &str,
            _args: &[String],
            _opts: Option<ExecStreamedOpts>,
        ) -> Result<(), ExecError> {
            unimplemented!("JobLifecycleRunner only supports run()")
        }
    }

    // --- Ported from client_test.go: TestGetVersion_* ---

    #[test]
    fn get_version_retries_transient_errors_until_reachable() {
        let mock = MockCommandRunner::new();
        let args = ["version", "-o", "json", "--request-timeout=5s"];

        // Two transient connection-refused failures, then success. Our
        // hand-rolled mock's `(name, args)` queue is FIFO per key, so
        // pushing these three expectations in order reproduces Go's
        // `.Times(2)` followed by `.Once()` sequencing without needing
        // call-count matchers.
        for _ in 0..2 {
            mock.expect_run(
                "kubectl",
                &args,
                ExecResult {
                    error: Some(ExecError::RunFailed(
                        "The connection to the server 127.0.0.1:6443 was refused - did you specify the right host or port?".to_string(),
                    )),
                    stderr: vec![
                        "The connection to the server 127.0.0.1:6443 was refused - did you specify the right host or port?"
                            .to_string(),
                    ],
                    ..Default::default()
                },
            );
        }

        mock.expect_run(
            "kubectl",
            &args,
            ExecResult {
                stdout: vec![r#"{"serverVersion":{"gitVersion":"v1.99.0+k3s1"}}"#.to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let got = client
            .get_version(Duration::from_secs(1), Duration::from_millis(1))
            .unwrap();
        assert_eq!(got.server_version.git_version, "v1.99.0+k3s1");
    }

    #[test]
    fn get_version_returns_permanent_errors_immediately() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["version", "-o", "json", "--request-timeout=5s"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                stderr: vec![
                    "error: You must be logged in to the server (Unauthorized)".to_string(),
                ],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client
            .get_version(Duration::from_secs(60), Duration::from_millis(1))
            .unwrap_err();
        assert!(err.to_string().contains("Unauthorized"));
    }

    #[test]
    fn get_version_times_out_while_api_stays_unreachable() {
        // Ported against `xianyun_testutil::MockCommandRunner::
        // expect_run_repeating`, the direct equivalent of Go's
        // `.Maybe()` (any call count, including zero) this crate's
        // shared mock now provides -- unblocking this test, previously
        // stubbed with a generous fixed-64-expectation stand-in. See
        // `PORT_NOTES.md`.
        let mock = MockCommandRunner::new();
        let args = ["version", "-o", "json", "--request-timeout=5s"];

        mock.expect_run_repeating("kubectl", &args, || ExecResult {
            error: Some(ExecError::RunFailed("kubectl failed".to_string())),
            stderr: vec!["The connection to the server 127.0.0.1:6443 was refused".to_string()],
            ..Default::default()
        });

        let client = new_client(mock);

        let err = client
            .get_version(Duration::from_millis(20), Duration::from_millis(1))
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("not ready"));
        assert!(msg.contains("refused"));
    }

    // --- Ported from client_test.go: TestGet_ErrorIncludesStderrTail ---

    #[test]
    fn get_error_includes_stderr_tail() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["get", "configmap", "-n", "kube-system", "-o", "json"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                stderr: vec![r#"Error from server (Forbidden): configmaps is forbidden: User "dev" cannot list resource "configmaps""#.to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client.get("configmap", "", "kube-system").unwrap_err();
        assert!(err.to_string().contains("Forbidden"));
    }

    // --- Ported from client_test.go: TestListResourceNames_* ---

    #[test]
    fn list_resource_names_empty() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "deployment",
                "-n",
                "cert-manager",
                "-l",
                "app.kubernetes.io/instance=cert-manager",
                "-o",
                "jsonpath={.items[*].metadata.name}",
            ],
            ExecResult {
                stdout: vec![String::new()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let names = client
            .list_resource_names(
                "deployment",
                "cert-manager",
                "app.kubernetes.io/instance=cert-manager",
            )
            .unwrap();
        assert!(names.is_empty());
    }

    #[test]
    fn list_resource_names_single() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "deployment",
                "-n",
                "cnpg-system",
                "-l",
                "app.kubernetes.io/instance=cloudnative-pg",
                "-o",
                "jsonpath={.items[*].metadata.name}",
            ],
            ExecResult {
                stdout: vec!["cloudnative-pg".to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let names = client
            .list_resource_names(
                "deployment",
                "cnpg-system",
                "app.kubernetes.io/instance=cloudnative-pg",
            )
            .unwrap();
        assert_eq!(names, vec!["cloudnative-pg".to_string()]);
    }

    #[test]
    fn list_resource_names_multiple() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "deployment",
                "-n",
                "cert-manager",
                "-l",
                "app.kubernetes.io/instance=cert-manager",
                "-o",
                "jsonpath={.items[*].metadata.name}",
            ],
            ExecResult {
                stdout: vec![
                    "cert-manager cert-manager-cainjector cert-manager-webhook".to_string(),
                ],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let names = client
            .list_resource_names(
                "deployment",
                "cert-manager",
                "app.kubernetes.io/instance=cert-manager",
            )
            .unwrap();
        assert_eq!(
            names,
            vec![
                "cert-manager".to_string(),
                "cert-manager-cainjector".to_string(),
                "cert-manager-webhook".to_string(),
            ]
        );
    }

    #[test]
    fn list_resource_names_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "deployment",
                "-n",
                "cnpg-system",
                "-l",
                "app.kubernetes.io/instance=cloudnative-pg",
                "-o",
                "jsonpath={.items[*].metadata.name}",
            ],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client
            .list_resource_names(
                "deployment",
                "cnpg-system",
                "app.kubernetes.io/instance=cloudnative-pg",
            )
            .unwrap_err();
        assert!(matches!(err, KubectlError::Exec(_)));
    }

    #[test]
    fn list_resource_names_empty_namespace() {
        // No mock expectations queued: the guard short-circuits before
        // any kubectl call.
        let mock = MockCommandRunner::new();
        let client = new_client(mock);

        let err = client
            .list_resource_names("deployment", "", "app.kubernetes.io/instance=cert-manager")
            .unwrap_err();
        assert!(matches!(err, KubectlError::NamespaceRequired));
    }

    // --- Ported from client_test.go: TestEnsureNamespace_* ---

    #[test]
    fn ensure_namespace() {
        let runner = StdinCapturingRunner::new(
            "kubectl",
            &["apply", "-f", "-"],
            r#"{"apiVersion":"v1","kind":"Namespace","metadata":{"name":"gitlab"}}"#,
            ExecResult::default(),
        );

        let client = Client::new(Arc::new(runner));

        client.ensure_namespace("gitlab").unwrap();
    }

    #[test]
    fn ensure_namespace_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["apply", "-f", "-"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client.ensure_namespace("gitlab").unwrap_err();
        assert!(err.to_string().contains("gitlab"));
    }

    #[test]
    fn ensure_namespace_error_includes_stderr_tail() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["apply", "-f", "-"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                stderr: vec![r#"Error from server (Forbidden): namespaces is forbidden: User "dev" cannot create resource "namespaces""#.to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client.ensure_namespace("gitlab").unwrap_err();
        assert!(err.to_string().contains("Forbidden"));
    }

    // --- Ported from client_test.go: TestDeleteNamespace_* ---

    #[test]
    fn delete_namespace() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["delete", "namespace", "gitlab", "--ignore-not-found"],
            ExecResult {
                stdout: vec![String::new()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        client.delete_namespace("gitlab").unwrap();
    }

    #[test]
    fn delete_namespace_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["delete", "namespace", "gitlab", "--ignore-not-found"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client.delete_namespace("gitlab").unwrap_err();
        assert!(err.to_string().contains("gitlab"));
    }

    // --- Ported from client_test.go: TestNamespaceExists_* ---

    #[test]
    fn namespace_exists_true() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "namespace",
                "gitlab",
                "--ignore-not-found",
                "-o",
                "name",
            ],
            ExecResult {
                stdout: vec!["namespace/gitlab".to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        assert!(client.namespace_exists("gitlab").unwrap());
    }

    #[test]
    fn namespace_exists_not_found() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "namespace",
                "gitlab",
                "--ignore-not-found",
                "-o",
                "name",
            ],
            ExecResult {
                stdout: vec![String::new()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        assert!(!client.namespace_exists("gitlab").unwrap());
    }

    #[test]
    fn namespace_exists_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "get",
                "namespace",
                "gitlab",
                "--ignore-not-found",
                "-o",
                "name",
            ],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client.namespace_exists("gitlab").unwrap_err();
        assert!(err.to_string().contains("gitlab"));
    }

    // --- Ported from client_test.go: TestWaitForNamespaceReady_* ---
    //
    // Go's `matchWaitArgs` helper asserts a `--timeout=<duration>` suffix
    // in `(0, budget]` rather than an exact value, because the elapsed
    // time between the two `kubectl wait` calls is real wall-clock time.
    // This crate's mock matches args exactly, so the ported tests below
    // assert against the *initial* full-budget timeout only; the
    // decreasing-remaining-budget behavior itself
    // (`wait_for_namespace_ready_total_duration_bounded`) needs a mock
    // that can match a `--timeout=` prefix rather than the exact
    // string -- flagged for a Phase 4 mock enhancement in PORT_NOTES.md.

    #[test]
    fn wait_for_namespace_ready() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "wait",
                "--for=jsonpath={.status.phase}=Active",
                "namespace/gitlab",
                "--timeout=30s",
            ],
            ExecResult {
                stdout: vec!["namespace/gitlab condition met".to_string()],
                ..Default::default()
            },
        );
        mock.expect_run(
            "kubectl",
            &[
                "wait",
                "--for=create",
                "serviceaccount/default",
                "-n",
                "gitlab",
                "--timeout=30s",
            ],
            ExecResult {
                stdout: vec!["serviceaccount/default condition met".to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        client
            .wait_for_namespace_ready("gitlab", Duration::from_secs(30))
            .unwrap();
    }

    #[test]
    fn wait_for_namespace_ready_phase_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "wait",
                "--for=jsonpath={.status.phase}=Active",
                "namespace/gitlab",
                "--timeout=30s",
            ],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client
            .wait_for_namespace_ready("gitlab", Duration::from_secs(30))
            .unwrap_err();
        assert!(err.to_string().contains("gitlab"));
    }

    #[test]
    fn wait_for_namespace_ready_service_account_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &[
                "wait",
                "--for=jsonpath={.status.phase}=Active",
                "namespace/gitlab",
                "--timeout=30s",
            ],
            ExecResult {
                stdout: vec!["namespace/gitlab condition met".to_string()],
                ..Default::default()
            },
        );
        mock.expect_run(
            "kubectl",
            &[
                "wait",
                "--for=create",
                "serviceaccount/default",
                "-n",
                "gitlab",
                "--timeout=30s",
            ],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client
            .wait_for_namespace_ready("gitlab", Duration::from_secs(30))
            .unwrap_err();
        assert!(err.to_string().contains("gitlab"));
    }

    #[test]
    fn wait_for_namespace_ready_total_duration_bounded() {
        // Ported against `xianyun_testutil::MockCommandRunner::
        // expect_run_matching`, the `mock.MatchedBy` equivalent
        // `PORT_NOTES.md` flagged this test as needing: rather than an
        // exact `--timeout=` string, each expectation's predicate
        // parses the trailing `--timeout=<N>s` argument and asserts the
        // remaining budget is in `(0, 30]` seconds, matching Go's
        // `matchWaitArgs` helper's decreasing-remaining-time assertion.
        fn matches_wait_args(prefix: &[&str], budget_secs: u64) -> impl Fn(&[String]) -> bool {
            let prefix: Vec<String> = prefix.iter().map(|s| s.to_string()).collect();
            move |args: &[String]| {
                if args.len() != prefix.len() + 1 {
                    return false;
                }
                if args[..prefix.len()] != prefix[..] {
                    return false;
                }

                let Some(timeout_str) = args[prefix.len()]
                    .strip_prefix("--timeout=")
                    .and_then(|s| s.strip_suffix('s'))
                else {
                    return false;
                };
                let Ok(remaining) = timeout_str.parse::<f64>() else {
                    return false;
                };

                remaining > 0.0 && remaining <= budget_secs as f64
            }
        }

        let mock = MockCommandRunner::new();
        mock.expect_run_matching(
            "kubectl",
            matches_wait_args(
                &[
                    "wait",
                    "--for=jsonpath={.status.phase}=Active",
                    "namespace/gitlab",
                ],
                30,
            ),
            ExecResult {
                stdout: vec!["namespace/gitlab condition met".to_string()],
                ..Default::default()
            },
        );
        mock.expect_run_matching(
            "kubectl",
            matches_wait_args(
                &[
                    "wait",
                    "--for=create",
                    "serviceaccount/default",
                    "-n",
                    "gitlab",
                ],
                30,
            ),
            ExecResult {
                stdout: vec!["serviceaccount/default condition met".to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        client
            .wait_for_namespace_ready("gitlab", Duration::from_secs(30))
            .unwrap();
    }

    // --- Ported from client_test.go: TestApply / TestDeleteManifest ---

    #[test]
    fn apply_with_namespace() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["apply", "-n", "gitlab", "-f", "-"],
            ExecResult::default(),
        );

        let client = new_client(mock);

        client
            .apply(b"apiVersion: v1\nkind: ConfigMap\n", "gitlab")
            .unwrap();
    }

    #[test]
    fn apply_without_namespace_uses_per_object_metadata() {
        let mock = MockCommandRunner::new();
        mock.expect_run("kubectl", &["apply", "-f", "-"], ExecResult::default());

        let client = new_client(mock);

        client
            .apply(b"apiVersion: v1\nkind: ConfigMap\n", "")
            .unwrap();
    }

    #[test]
    fn delete_manifest_with_namespace() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["delete", "-n", "gitlab", "-f", "-", "--ignore-not-found"],
            ExecResult::default(),
        );

        let client = new_client(mock);

        client
            .delete_manifest(b"apiVersion: v1\nkind: ConfigMap\n", "gitlab")
            .unwrap();
    }

    #[test]
    fn delete_manifest_without_namespace_uses_per_object_metadata() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["delete", "-f", "-", "--ignore-not-found"],
            ExecResult::default(),
        );

        let client = new_client(mock);

        client
            .delete_manifest(b"apiVersion: v1\nkind: ConfigMap\n", "")
            .unwrap();
    }

    #[test]
    fn delete_manifest_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["delete", "-n", "gitlab", "-f", "-", "--ignore-not-found"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client.delete_manifest(b"invalid", "gitlab").unwrap_err();
        assert!(matches!(err, KubectlError::Exec(_)));
    }

    // --- Ported from client_test.go: TestGetRaw / TestGetRaw_Error ---

    #[test]
    fn get_raw() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["get", "--raw", "/api/v1/nodes/node-a/proxy/configz"],
            ExecResult {
                stdout: vec![r#"{"kubeletconfig":{"clusterDomain":"cluster.local"}}"#.to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let out = client
            .get_raw("/api/v1/nodes/node-a/proxy/configz")
            .unwrap();
        let got: serde_json::Value = serde_json::from_str(&out).unwrap();
        let want: serde_json::Value =
            serde_json::from_str(r#"{"kubeletconfig":{"clusterDomain":"cluster.local"}}"#).unwrap();
        assert_eq!(got, want);
    }

    #[test]
    fn get_raw_error() {
        let mock = MockCommandRunner::new();
        mock.expect_run(
            "kubectl",
            &["get", "--raw", "/api/v1/nodes/node-a/proxy/configz"],
            ExecResult {
                error: Some(ExecError::RunFailed("kubectl failed".to_string())),
                stderr: vec![r#"Error from server (Forbidden): nodes "node-a" is forbidden: User "dev" cannot get resource "nodes/proxy""#.to_string()],
                ..Default::default()
            },
        );

        let client = new_client(mock);

        let err = client
            .get_raw("/api/v1/nodes/node-a/proxy/configz")
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("/api/v1/nodes/node-a/proxy/configz"));
        assert!(msg.contains("Forbidden"));
    }

    // --- Ported from client_test.go: TestRunJob_* ---
    //
    // `probeJobManifest` mirrors Go's local struct used to decode the
    // Job manifest RunJob writes to stdin. Go's tests capture it via
    // `mock.RunAndReturn` reading `opts.Stdin`; this crate's mock does
    // not thread `Options`/stdin through expectations at all (see
    // `testutil.rs`), so these three are stubbed with `unimplemented!()`
    // rather than mechanically ported against a mock that cannot express
    // "read what was written, then assert on it and use it to build the
    // next expected args". Un-ignoring these in Phase 4 needs a
    // `RunAndReturn`-equivalent mock capability. See PORT_NOTES.md.

    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct ProbeJobManifest {
        #[serde(rename = "apiVersion")]
        api_version: String,
        kind: String,
        metadata: ProbeJobMetadata,
        spec: ProbeJobSpec,
    }

    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct ProbeJobMetadata {
        name: String,
        namespace: String,
    }

    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct ProbeJobSpec {
        #[serde(rename = "backoffLimit")]
        backoff_limit: Option<i64>,
        #[serde(rename = "ttlSecondsAfterFinished")]
        ttl_seconds_after_finished: Option<i64>,
        #[serde(rename = "activeDeadlineSeconds")]
        active_deadline_seconds: i64,
        template: ProbeJobTemplate,
    }

    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct ProbeJobTemplate {
        spec: ProbeJobTemplateSpec,
    }

    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct ProbeJobTemplateSpec {
        #[serde(rename = "restartPolicy")]
        restart_policy: String,
        containers: Vec<ProbeJobContainer>,
    }

    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct ProbeJobContainer {
        image: String,
        command: Vec<String>,
    }

    #[test]
    fn run_job_returns_logs_and_deletes_job() {
        let runner = JobLifecycleRunner::new(
            "default",
            vec![r#"{"status":{"conditions":[{"type":"Complete","status":"True"}]}}"#],
            Some(vec![
                "search default.svc.cluster.local svc.cluster.local cluster.local",
                "nameserver 10.43.0.10",
            ]),
        );

        let runner = Arc::new(runner);
        let client = Client::new(runner.clone());

        let out = client
            .run_job(
                "busybox:1.36",
                "default",
                Duration::from_secs(60),
                &["cat", "/etc/resolv.conf"],
            )
            .unwrap();

        assert_eq!(
            out,
            "search default.svc.cluster.local svc.cluster.local cluster.local\nnameserver 10.43.0.10"
        );
        assert_eq!(
            runner.delete_call_count(),
            1,
            "the probe Job must be deleted"
        );
    }

    #[test]
    fn run_job_failed_job_reports_reason_without_waiting_for_timeout() {
        let runner = JobLifecycleRunner::new(
            "default",
            vec![
                r#"{"status":{"conditions":[{"type":"Failed","status":"True","reason":"BackoffLimitExceeded","message":"Job has reached the specified backoff limit"}]}}"#,
            ],
            None,
        );

        let start = std::time::Instant::now();
        let runner = Arc::new(runner);
        let client = Client::new(runner.clone());

        let err = client
            .run_job(
                "busybox:1.36",
                "default",
                Duration::from_secs(3600),
                &["cat", "/etc/resolv.conf"],
            )
            .unwrap_err();

        assert!(err.to_string().contains("BackoffLimitExceeded"));
        assert!(
            err.to_string()
                .contains("Job has reached the specified backoff limit")
        );
        assert!(
            start.elapsed() < Duration::from_secs(10),
            "a failed Job must be reported as soon as it is observed"
        );
        assert_eq!(
            runner.delete_call_count(),
            1,
            "the probe Job must be deleted"
        );
    }

    #[test]
    fn run_job_deletes_job_when_deadline_elapses() {
        // The Job never reaches a terminal condition (e.g.
        // ImagePullBackOff); `get job` is polled an indeterminate
        // number of times (Go's `.Maybe()`) before the deadline elapses.
        let runner = JobLifecycleRunner::repeating("default", r#"{"status":{}}"#);
        let runner = Arc::new(runner);
        let client = Client::new(runner.clone());

        let err = client
            .run_job(
                "busybox:1.36",
                "default",
                Duration::from_millis(20),
                &["cat", "/etc/resolv.conf"],
            )
            .unwrap_err();

        assert!(err.to_string().contains("deadline exceeded"));
        assert_eq!(
            runner.delete_call_count(),
            1,
            "cleanup must still run when the deadline elapses"
        );
    }
}