xianyun_deploy/reloader/mirror/

copy.rs

//! Prepares and cleans up `-caproni` deployment copies used to run mirror
//! reloaders without disrupting the original workload.
//!
//! Ported from `internal/reloader/mirror/copy.go`.

use std::collections::HashMap;
use std::time::Duration;

use serde_json::Value;
use thiserror::Error;
use xianyun_core::config::latest::ReloaderConfig;

use super::manager::Manager;

/// The suffix appended to the original deployment name to form the name
/// of the mirror copy managed by caproni.
pub const COPY_SUFFIX: &str = "-caproni";

/// The number of replicas the original deployment will be scaled up to.
/// This is intentionally hard-coded to avoid storing state in the old
/// deployments or within caproni.
///
/// See <https://gitlab.com/gitlab-org/gitlab/-/issues/590495#note_3118758329>
pub const SCALE_UP_REPLICA_COUNT: i64 = 1;

/// Errors from deployment-copy preparation and cleanup.
#[derive(Debug, Error)]
pub enum CopyError {
    /// A deployment JSON structure was invalid in a way that prevented
    /// modifying it for caproni use.
    #[error("invalid deployment structure: {0}")]
    InvalidDeploymentStructure(String),
    /// A specified container was not found in the deployment.
    #[error("container not found in deployment: {0}")]
    ContainerNotFound(String),
    /// The same container name appeared more than once across multiple
    /// reloaders targeting the same pod, or within one `files:`-style
    /// container name list.
    #[error("duplicate container for target: {0}")]
    DuplicateContainer(String),
    /// One reloader targets an entire pod and another targets a
    /// specific container within the same pod.
    #[error("cannot mix pod-level and container-level targets for the same resource: {0}")]
    MixedPodAndContainerTargets(String),
    /// Validating a reloader's mirror target failed.
    #[error("validating mirror target for reloader {0:?}: {1}")]
    ValidatingTarget(String, #[source] Box<CopyError>),
    /// Fetching, applying, scaling, deleting, or waiting on a resource
    /// via the kubectl backend failed.
    #[error("{0}")]
    Kubectl(String, #[source] xianyun_backends::kubectl::KubectlError),
    /// Parsing a fetched deployment's JSON failed.
    #[error("parsing deployment JSON for {0}: {1}")]
    ParseJson(String, #[source] serde_json::Error),
    /// Serializing a modified deployment back to JSON failed.
    #[error("marshaling modified deployment {0}: {1}")]
    SerializeJson(String, #[source] serde_json::Error),
}

/// A Kubernetes resource to be mirrored.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Target {
    pub namespace: String,
    pub kubernetes_type: String,
    pub name: String,
    /// Optional. When non-empty, only the listed containers are
    /// replaced in the deployment copy. When empty, all containers are
    /// replaced.
    pub containers: Vec<String>,
    /// `true` if one reloader target was specified without a container
    /// name. It is only an intermediate value; outside consumers will
    /// find an empty `containers` if this was `true` at some point.
    has_wildcard: bool,
}

impl Target {
    /// Creates a new [`Target`] with no containers set. A convenience
    /// constructor for callers outside this module, since `has_wildcard`
    /// is a private intermediate field that struct-update syntax
    /// (`..Default::default()`) cannot reach across module boundaries.
    #[must_use]
    pub fn new(namespace: &str, kubernetes_type: &str, name: &str) -> Self {
        Self {
            namespace: namespace.to_string(),
            kubernetes_type: kubernetes_type.to_string(),
            name: name.to_string(),
            containers: Vec::new(),
            has_wildcard: false,
        }
    }

    /// Merges one reloader's container target into this entry,
    /// validating for conflicts.
    fn add_container(&mut self, container: &str) -> Result<(), CopyError> {
        let resource_id = format!("{}/{}/{}", self.namespace, self.kubernetes_type, self.name);

        if container.is_empty() {
            // Pod-level target: cannot mix with container-level targets.
            if !self.containers.is_empty() {
                return Err(CopyError::MixedPodAndContainerTargets(resource_id));
            }

            self.has_wildcard = true;

            return Ok(());
        }

        // Container-level target: cannot mix with pod-level targets.
        if self.has_wildcard {
            return Err(CopyError::MixedPodAndContainerTargets(resource_id));
        }

        if self.containers.iter().any(|c| c == container) {
            return Err(CopyError::DuplicateContainer(format!(
                "{container} in {resource_id}"
            )));
        }

        self.containers.push(container.to_string());

        Ok(())
    }
}

/// Holds information about a prepared deployment copy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CopyInfo {
    pub resource_type: String,
    pub name: String,
    pub namespace: String,
}

/// Collects mirror targets from all reloaders, merges targets for the
/// same pod, and validates that there are no conflicts. Returns a
/// deduplicated, sorted list of targets ready for use with
/// [`Manager::prepare_copies`].
///
/// Validation rules:
/// - The same container may not appear in more than one reloader for
///   the same pod.
/// - A pod-level target (no container specified) cannot be combined
///   with a container-level target for the same pod.
///
/// # Errors
///
/// Returns [`CopyError::DuplicateContainer`] or
/// [`CopyError::MixedPodAndContainerTargets`] (wrapped in
/// [`CopyError::ValidatingTarget`]) if a reloader's target conflicts
/// with another's.
pub fn collect_and_validate_targets(
    reloaders: &HashMap<String, ReloaderConfig>,
) -> Result<Vec<Target>, CopyError> {
    let mut merged: HashMap<String, Target> = HashMap::new();
    // Preserves first-seen order per key, matching the Go original's
    // `slices.Sorted(maps.Keys(merged))` -- sorted lexicographically by
    // the "namespace/type/name" key, which is what determines the
    // output order regardless of (randomized) map iteration order.
    let mut order: Vec<String> = Vec::new();

    for (name, reloader_config) in reloaders {
        if reloader_config.reloader_type != "mirror" {
            continue;
        }

        let Some(mirror) = reloader_config.mirror.as_ref() else {
            continue;
        };
        let Some(target) = mirror.target.as_ref() else {
            continue;
        };

        let key = format!(
            "{}/{}/{}",
            target.namespace, target.kubernetes_type, target.name
        );

        let entry = merged.entry(key.clone()).or_insert_with(|| {
            order.push(key.clone());
            Target {
                namespace: target.namespace.clone(),
                kubernetes_type: target.kubernetes_type.clone(),
                name: target.name.clone(),
                containers: Vec::new(),
                has_wildcard: false,
            }
        });

        entry
            .add_container(target.container.trim())
            .map_err(|e| CopyError::ValidatingTarget(name.clone(), Box::new(e)))?;
    }

    if merged.is_empty() {
        return Ok(Vec::new());
    }

    order.sort();

    let mut targets = Vec::with_capacity(order.len());

    for key in order {
        let mut entry = merged.remove(&key).expect("key came from merged");
        entry.containers.sort();
        entry.has_wildcard = false; // clear intermediate value
        targets.push(entry);
    }

    Ok(targets)
}

/// Modifies a deployment's JSON structure for use as a `-caproni` copy.
/// Renames the deployment, removes health probes, and replaces
/// container commands with `sleep infinity`. When `containers` is
/// non-empty, only those containers are modified; all others are left
/// unchanged.
///
/// # Errors
///
/// Returns [`CopyError::InvalidDeploymentStructure`] if the JSON does
/// not have the expected shape, or [`CopyError::ContainerNotFound`] if
/// a named container in `containers` does not exist in the deployment.
pub fn modify_deployment_for_caproni(
    deployment: &mut Value,
    target_name: &str,
    image: &str,
    containers: &[String],
) -> Result<(), CopyError> {
    let metadata = deployment
        .get_mut("metadata")
        .and_then(Value::as_object_mut)
        .ok_or_else(|| {
            CopyError::InvalidDeploymentStructure("metadata is not an object".to_string())
        })?;

    metadata.insert(
        "name".to_string(),
        Value::String(format!("{target_name}{COPY_SUFFIX}")),
    );

    let container_set = build_container_set(containers)?;

    let container_list = extract_container_list_mut(deployment)?;

    let mut matched: HashMap<String, bool> = HashMap::new();

    for container in container_list.iter_mut() {
        let c = container.as_object_mut().ok_or_else(|| {
            CopyError::InvalidDeploymentStructure("container is not an object".to_string())
        })?;

        let container_name = c
            .get("name")
            .and_then(Value::as_str)
            .filter(|s| !s.is_empty())
            .ok_or_else(|| {
                CopyError::InvalidDeploymentStructure("container has no name".to_string())
            })?
            .to_string();

        if !containers.is_empty() && !container_set.contains_key(&container_name) {
            continue;
        }

        matched.insert(container_name, true);

        c.remove("livenessProbe");
        c.remove("readinessProbe");
        c.remove("startupProbe");
        c.insert(
            "command".to_string(),
            Value::Array(vec![
                Value::String("sleep".to_string()),
                Value::String("infinity".to_string()),
            ]),
        );
        c.insert("image".to_string(), Value::String(image.to_string()));
    }

    for name in containers {
        if !matched.contains_key(name) {
            return Err(CopyError::ContainerNotFound(name.clone()));
        }
    }

    Ok(())
}

/// Traverses the deployment JSON structure and returns a mutable
/// reference to the containers array. Ported from `extractContainerList`,
/// specialized to the one fixed path this module ever reads (the Go
/// original's general-purpose `dyn.Access` is not otherwise used by
/// this crate, so it was not ported wholesale).
fn extract_container_list_mut(deployment: &mut Value) -> Result<&mut Vec<Value>, CopyError> {
    let spec = get_or_not_found(deployment, "spec", "spec")?;
    let template = get_or_not_found(spec, "template", "spec.template")?;
    let template_spec = get_or_not_found(template, "spec", "spec.template.spec")?;

    template_spec
        .get_mut("containers")
        .and_then(Value::as_array_mut)
        .ok_or_else(|| {
            CopyError::InvalidDeploymentStructure(
                "spec.template.spec.containers: not found".to_string(),
            )
        })
}

fn get_or_not_found<'a>(
    value: &'a mut Value,
    key: &str,
    path_so_far: &str,
) -> Result<&'a mut Value, CopyError> {
    value
        .get_mut(key)
        .ok_or_else(|| CopyError::InvalidDeploymentStructure(format!("{path_so_far}: not found")))
}

/// Validates that `containers` has no duplicates and returns a set for
/// O(1) lookup.
fn build_container_set(containers: &[String]) -> Result<HashMap<String, bool>, CopyError> {
    let mut set = HashMap::with_capacity(containers.len());

    for name in containers {
        if set.contains_key(name) {
            return Err(CopyError::DuplicateContainer(name.clone()));
        }

        set.insert(name.clone(), true);
    }

    Ok(set)
}

impl Manager {
    /// Scales down original deployments and creates `-caproni` copies
    /// for the given targets. Skips unsupported resource types.
    ///
    /// Callers are responsible for passing a deduplicated, pre-merged
    /// target list; in practice, callers should use
    /// [`collect_and_validate_targets`] to produce it. Returns a list
    /// of [`CopyInfo`] for cleanup operations.
    ///
    /// # Errors
    ///
    /// Returns [`CopyError`] if scaling down, fetching, modifying, or
    /// applying a copy fails for any target.
    pub fn prepare_copies(
        &self,
        targets: &[Target],
        copy_image: &str,
    ) -> Result<Vec<CopyInfo>, CopyError> {
        let mut infos = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();

        for target in targets {
            if target.kubernetes_type != "deployment" && target.kubernetes_type != "statefulset" {
                tracing::warn!(
                    r#type = %target.kubernetes_type,
                    name = %target.name,
                    namespace = %target.namespace,
                    "skipping copy-creation for target with unsupported resource type"
                );
                continue;
            }

            let key = format!(
                "{}/{}/{}",
                target.namespace, target.kubernetes_type, target.name
            );
            if !seen.insert(key) {
                continue;
            }

            tracing::info!(
                r#type = %target.kubernetes_type,
                name = %target.name,
                namespace = %target.namespace,
                "preparing deployment copy"
            );

            self.scale_down(target)?;
            self.create_caproni_deployment(target, copy_image)?;

            infos.push(CopyInfo {
                resource_type: target.kubernetes_type.clone(),
                name: target.name.clone(),
                namespace: target.namespace.clone(),
            });
        }

        Ok(infos)
    }

    /// Scales the original deployment to 0 replicas.
    fn scale_down(&self, target: &Target) -> Result<(), CopyError> {
        tracing::info!(
            r#type = %target.kubernetes_type,
            name = %target.name,
            namespace = %target.namespace,
            "scaling down original deployment"
        );

        self.kubectl
            .scale(&target.kubernetes_type, &target.name, &target.namespace, 0)
            .map_err(|e| {
                CopyError::Kubectl(
                    format!(
                        "scaling down {}/{}: {e}",
                        target.kubernetes_type, target.name
                    ),
                    e,
                )
            })
    }

    /// Fetches the original deployment, modifies it for caproni use,
    /// applies it as a `-caproni` deployment, scales it up, and waits
    /// for it to become available.
    fn create_caproni_deployment(
        &self,
        target: &Target,
        copy_image: &str,
    ) -> Result<(), CopyError> {
        let deployment = self.fetch_and_modify_deployment(target, copy_image)?;
        self.apply_and_scale_caproni_deployment(target, &deployment)
    }

    /// Fetches the original deployment JSON and modifies it for
    /// caproni use.
    fn fetch_and_modify_deployment(
        &self,
        target: &Target,
        copy_image: &str,
    ) -> Result<Value, CopyError> {
        tracing::info!(
            r#type = %target.kubernetes_type,
            name = %target.name,
            namespace = %target.namespace,
            "fetching deployment JSON"
        );

        let json_data = self
            .kubectl
            .get(&target.kubernetes_type, &target.name, &target.namespace)
            .map_err(|e| {
                CopyError::Kubectl(
                    format!("getting {}/{}: {e}", target.kubernetes_type, target.name),
                    e,
                )
            })?;

        let mut deployment: Value = serde_json::from_str(&json_data).map_err(|e| {
            CopyError::ParseJson(format!("{}/{}", target.kubernetes_type, target.name), e)
        })?;

        modify_deployment_for_caproni(
            &mut deployment,
            &target.name,
            copy_image,
            &target.containers,
        )?;

        Ok(deployment)
    }

    /// Applies the modified deployment, scales it up, and waits for
    /// availability.
    fn apply_and_scale_caproni_deployment(
        &self,
        target: &Target,
        deployment: &Value,
    ) -> Result<(), CopyError> {
        let modified_json = serde_json::to_vec(deployment).map_err(|e| {
            CopyError::SerializeJson(format!("{}/{}", target.kubernetes_type, target.name), e)
        })?;

        let copy_name = format!("{}{COPY_SUFFIX}", target.name);

        tracing::info!(
            r#type = %target.kubernetes_type,
            name = %copy_name,
            namespace = %target.namespace,
            "creating copy deployment"
        );

        self.kubectl
            .apply(&modified_json, &target.namespace)
            .map_err(|e| CopyError::Kubectl(format!("applying copy deployment: {e}"), e))?;

        tracing::info!(
            r#type = %target.kubernetes_type,
            name = %copy_name,
            namespace = %target.namespace,
            "scaling up copy deployment"
        );

        self.kubectl
            .scale(&target.kubernetes_type, &copy_name, &target.namespace, 1)
            .map_err(|e| CopyError::Kubectl(format!("scaling copy deployment: {e}"), e))?;

        tracing::info!(
            r#type = %target.kubernetes_type,
            name = %copy_name,
            namespace = %target.namespace,
            "waiting for copy rollout to complete"
        );

        self.kubectl
            .wait_for_rollout_status(
                &target.kubernetes_type,
                &copy_name,
                &target.namespace,
                Duration::from_secs(3 * 60),
            )
            .map_err(|e| CopyError::Kubectl(format!("waiting for copy rollout: {e}"), e))
    }

    /// Deletes a `-caproni` deployment copy and scales the original
    /// back up.
    ///
    /// # Errors
    ///
    /// Returns [`CopyError`] if deleting, scaling, or waiting fails.
    pub fn cleanup_copy(&self, deployment: &CopyInfo) -> Result<(), CopyError> {
        let copy_name = format!("{}{COPY_SUFFIX}", deployment.name);

        tracing::info!(
            r#type = %deployment.resource_type,
            name = %copy_name,
            namespace = %deployment.namespace,
            "deleting copy deployment"
        );

        self.kubectl
            .delete(&deployment.resource_type, &copy_name, &deployment.namespace)
            .map_err(|e| {
                CopyError::Kubectl(
                    format!(
                        "deleting {}/{}{COPY_SUFFIX} in {}: {e}",
                        deployment.resource_type, deployment.name, deployment.namespace
                    ),
                    e,
                )
            })?;

        tracing::info!(
            r#type = %deployment.resource_type,
            name = %deployment.name,
            namespace = %deployment.namespace,
            "scaling up original deployment"
        );

        self.kubectl
            .scale(
                &deployment.resource_type,
                &deployment.name,
                &deployment.namespace,
                SCALE_UP_REPLICA_COUNT,
            )
            .map_err(|e| {
                CopyError::Kubectl(
                    format!(
                        "scaling up {}/{} in {}: {e}",
                        deployment.resource_type, deployment.name, deployment.namespace
                    ),
                    e,
                )
            })?;

        tracing::info!(
            r#type = %deployment.resource_type,
            name = %deployment.name,
            namespace = %deployment.namespace,
            "waiting for original deployment to be available"
        );

        self.kubectl
            .wait(
                &deployment.resource_type,
                &deployment.name,
                &deployment.namespace,
                "available",
                Duration::from_secs(60),
            )
            .map_err(|e| {
                CopyError::Kubectl(
                    format!(
                        "waiting for {}/{} in {}: {e}",
                        deployment.resource_type, deployment.name, deployment.namespace
                    ),
                    e,
                )
            })
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use serde_json::json;
    use xianyun_backends::kubectl::KubectlError;
    use xianyun_core::config::latest::{MirrorConfig, MirrorTargetConfig, ReloaderConfig};

    use super::super::manager::Manager;
    use super::super::test_support::MockKubectlBackend;
    use super::*;

    const TEST_COPY_IMAGE: &str = "docker.io/library/alpine:test";

    fn valid_deployment_json(name: &str) -> String {
        json!({
            "metadata": {"name": name},
            "spec": {
                "selector": {"matchLabels": {"app": name}},
                "template": {
                    "spec": {
                        "containers": [{
                            "name": "app",
                            "image": "nginx:latest",
                            "command": ["nginx"],
                            "livenessProbe": {"httpGet": {"port": 80}},
                            "readinessProbe": {"httpGet": {"port": 80}},
                            "startupProbe": {"httpGet": {"port": 80}}
                        }]
                    }
                }
            }
        })
        .to_string()
    }

    fn kubectl_err(msg: &str) -> KubectlError {
        KubectlError::JobFailed(msg.to_string())
    }

    // Ported from internal/reloader/mirror/copy_test.go:58 (TestModifyDeploymentForCaproni)
    #[test]
    fn test_modify_deployment_for_caproni() {
        struct Case {
            name: &'static str,
            input: &'static str,
            container_names: Vec<&'static str>,
            expected: Option<serde_json::Value>,
            want_err: Option<&'static str>,
        }

        let cases = vec![
            Case {
                name: "single container",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "command": ["start"], "livenessProbe": {}, "readinessProbe": {}}
                    ]}}}
                }"#,
                container_names: vec![],
                expected: Some(json!({
                    "metadata": {"name": "myapp-caproni"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE}
                    ]}}}
                })),
                want_err: None,
            },
            Case {
                name: "all probes removed including startupProbe",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "command": ["start"], "livenessProbe": {}, "readinessProbe": {}, "startupProbe": {"httpGet": {"port": 8080}}}
                    ]}}}
                }"#,
                container_names: vec![],
                expected: Some(json!({
                    "metadata": {"name": "myapp-caproni"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE}
                    ]}}}
                })),
                want_err: None,
            },
            Case {
                name: "multiple containers",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app1", "livenessProbe": {}},
                        {"name": "app2", "readinessProbe": {}}
                    ]}}}
                }"#,
                container_names: vec![],
                expected: Some(json!({
                    "metadata": {"name": "myapp-caproni"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app1", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE},
                        {"name": "app2", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE}
                    ]}}}
                })),
                want_err: None,
            },
            Case {
                name: "named container - only target container replaced",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app1", "image": "nginx:latest", "livenessProbe": {}},
                        {"name": "app2", "image": "redis:latest", "readinessProbe": {}}
                    ]}}}
                }"#,
                container_names: vec!["app1"],
                expected: Some(json!({
                    "metadata": {"name": "myapp-caproni"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app1", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE},
                        {"name": "app2", "image": "redis:latest", "readinessProbe": {}}
                    ]}}}
                })),
                want_err: None,
            },
            Case {
                name: "multiple named containers - both replaced",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app1", "image": "nginx:latest", "livenessProbe": {}},
                        {"name": "app2", "image": "redis:latest", "readinessProbe": {}}
                    ]}}}
                }"#,
                container_names: vec!["app1", "app2"],
                expected: Some(json!({
                    "metadata": {"name": "myapp-caproni"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app1", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE},
                        {"name": "app2", "command": ["sleep", "infinity"], "image": TEST_COPY_IMAGE}
                    ]}}}
                })),
                want_err: None,
            },
            Case {
                name: "named container - not found returns error",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "image": "nginx:latest"}
                    ]}}}
                }"#,
                container_names: vec!["doesnotexist"],
                expected: None,
                want_err: Some("container not found in deployment: doesnotexist"),
            },
            Case {
                name: "one of two named containers not found returns error",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "image": "nginx:latest"}
                    ]}}}
                }"#,
                container_names: vec!["app", "doesnotexist"],
                expected: None,
                want_err: Some("container not found in deployment: doesnotexist"),
            },
            Case {
                name: "missing metadata",
                input: "{}",
                container_names: vec![],
                expected: None,
                want_err: Some("invalid deployment structure: metadata is not an object"),
            },
            Case {
                name: "missing spec",
                input: r#"{"metadata": {"name": "myapp"}}"#,
                container_names: vec![],
                expected: None,
                want_err: Some("invalid deployment structure: spec: not found"),
            },
            Case {
                name: "missing template",
                input: r#"{"metadata": {"name": "myapp"}, "spec": {}}"#,
                container_names: vec![],
                expected: None,
                want_err: Some("invalid deployment structure: spec.template: not found"),
            },
            Case {
                name: "missing template spec",
                input: r#"{"metadata": {"name": "myapp"}, "spec": {"template": {}}}"#,
                container_names: vec![],
                expected: None,
                want_err: Some("invalid deployment structure: spec.template.spec: not found"),
            },
            Case {
                name: "missing containers",
                input: r#"{"metadata": {"name": "myapp"}, "spec": {"template": {"spec": {}}}}"#,
                container_names: vec![],
                expected: None,
                want_err: Some(
                    "invalid deployment structure: spec.template.spec.containers: not found",
                ),
            },
            Case {
                name: "invalid container",
                input: r#"{"metadata": {"name": "myapp"}, "spec": {"template": {"spec": {"containers": ["not-a-map"]}}}}"#,
                container_names: vec![],
                expected: None,
                want_err: Some("invalid deployment structure: container is not an object"),
            },
            Case {
                name: "container missing name field returns error",
                input: r#"{"metadata": {"name": "myapp"}, "spec": {"template": {"spec": {"containers": [{"image": "nginx:latest"}]}}}}"#,
                container_names: vec![],
                expected: None,
                want_err: Some("invalid deployment structure: container has no name"),
            },
            Case {
                name: "duplicate container name in containers slice returns error",
                input: r#"{
                    "metadata": {"name": "myapp"},
                    "spec": {"template": {"spec": {"containers": [
                        {"name": "app", "image": "nginx:latest"}
                    ]}}}
                }"#,
                container_names: vec!["app", "app"],
                expected: None,
                want_err: Some("duplicate container for target: app"),
            },
        ];

        for case in cases {
            let mut deployment: Value = serde_json::from_str(case.input).unwrap();
            let container_names: Vec<String> =
                case.container_names.iter().map(|s| s.to_string()).collect();

            let result = modify_deployment_for_caproni(
                &mut deployment,
                "myapp",
                TEST_COPY_IMAGE,
                &container_names,
            );

            if let Some(want_err) = case.want_err {
                let err = result.unwrap_err();
                assert!(
                    err.to_string().contains(want_err),
                    "{}: expected error containing {want_err:?}, got {err}",
                    case.name
                );
                continue;
            }

            result.unwrap_or_else(|e| panic!("{}: unexpected error: {e}", case.name));
            assert_eq!(deployment, case.expected.unwrap(), "{}", case.name);
        }
    }

    #[derive(Default)]
    struct CopyMocksData {
        get_json: Option<String>,
        scale_err: bool,
        get_err: bool,
        apply_err: bool,
        scale_up_err: bool,
        wait_err: bool,
    }

    fn setup_copy_mocks(
        mut builder: MockKubectlBackend,
        target: &Target,
        data: &CopyMocksData,
    ) -> MockKubectlBackend {
        let name = target.name.clone();
        let scale_err = data.scale_err;

        builder = builder.expect_scale(move |_res_type, _name, _ns, replicas| {
            assert_eq!(replicas, 0);
            if scale_err {
                Err(kubectl_err("scale failed"))
            } else {
                Ok(())
            }
        });

        if data.scale_err {
            return builder;
        }

        let get_json = data
            .get_json
            .clone()
            .unwrap_or_else(|| valid_deployment_json(&name));
        let get_err = data.get_err;
        let is_valid_json = serde_json::from_str::<Value>(&get_json).is_ok();

        builder = builder.expect_get(move |_res_type, _name, _ns| {
            if get_err {
                Err(kubectl_err("get failed"))
            } else {
                Ok(get_json.clone())
            }
        });

        if data.get_err || !is_valid_json {
            return builder;
        }

        let apply_err = data.apply_err;
        builder = builder.expect_apply(move |_json, _ns| {
            if apply_err {
                Err(kubectl_err("apply failed"))
            } else {
                Ok(())
            }
        });

        if data.apply_err {
            return builder;
        }

        let scale_up_err = data.scale_up_err;
        builder = builder.expect_scale(move |_res_type, _name, _ns, replicas| {
            assert_eq!(replicas, 1);
            if scale_up_err {
                Err(kubectl_err("scale failed"))
            } else {
                Ok(())
            }
        });

        if data.scale_up_err {
            return builder;
        }

        let wait_err = data.wait_err;
        builder.expect_wait_for_rollout_status(move |_res_type, _name, _ns| {
            if wait_err {
                Err(kubectl_err("wait failed"))
            } else {
                Ok(())
            }
        })
    }

    fn setup_deduplicated_copy_mocks(
        mut builder: MockKubectlBackend,
        targets: &[Target],
        data: &CopyMocksData,
    ) -> MockKubectlBackend {
        let mut seen = std::collections::HashSet::new();

        for target in targets {
            let key = format!(
                "{}/{}/{}",
                target.namespace, target.kubernetes_type, target.name
            );
            if !seen.insert(key) {
                continue;
            }

            if target.kubernetes_type == "deployment" || target.kubernetes_type == "statefulset" {
                builder = setup_copy_mocks(builder, target, data);
            }
        }

        builder
    }

    // Ported from internal/reloader/mirror/copy_test.go:309 (TestPrepareCopies)
    #[test]
    fn test_prepare_copies() {
        struct Case {
            name: &'static str,
            targets: Vec<Target>,
            mock_data: CopyMocksData,
            want_infos: Vec<CopyInfo>,
            want_err: Option<&'static str>,
        }

        fn target(namespace: &str, kubernetes_type: &str, name: &str) -> Target {
            Target {
                namespace: namespace.to_string(),
                kubernetes_type: kubernetes_type.to_string(),
                name: name.to_string(),
                containers: Vec::new(),
                ..Default::default()
            }
        }

        fn info(resource_type: &str, name: &str, namespace: &str) -> CopyInfo {
            CopyInfo {
                resource_type: resource_type.to_string(),
                name: name.to_string(),
                namespace: namespace.to_string(),
            }
        }

        let cases = vec![
            Case {
                name: "success with deployment",
                targets: vec![target("default", "deployment", "myapp")],
                mock_data: CopyMocksData::default(),
                want_infos: vec![info("deployment", "myapp", "default")],
                want_err: None,
            },
            Case {
                name: "success with statefulset",
                targets: vec![target("prod", "statefulset", "mydb")],
                mock_data: CopyMocksData::default(),
                want_infos: vec![info("statefulset", "mydb", "prod")],
                want_err: None,
            },
            Case {
                name: "skips unsupported type",
                targets: vec![target("default", "daemonset", "myapp")],
                mock_data: CopyMocksData::default(),
                want_infos: vec![],
                want_err: None,
            },
            Case {
                name: "deduplicates targets",
                targets: vec![
                    target("default", "deployment", "myapp"),
                    target("default", "deployment", "myapp"),
                ],
                mock_data: CopyMocksData::default(),
                want_infos: vec![info("deployment", "myapp", "default")],
                want_err: None,
            },
            Case {
                name: "scale down error",
                targets: vec![target("default", "deployment", "myapp")],
                mock_data: CopyMocksData {
                    scale_err: true,
                    ..Default::default()
                },
                want_infos: vec![],
                want_err: Some("scale failed"),
            },
            Case {
                name: "get error",
                targets: vec![target("default", "deployment", "myapp")],
                mock_data: CopyMocksData {
                    get_err: true,
                    ..Default::default()
                },
                want_infos: vec![],
                want_err: Some("get failed"),
            },
            Case {
                name: "apply error",
                targets: vec![target("default", "deployment", "myapp")],
                mock_data: CopyMocksData {
                    apply_err: true,
                    ..Default::default()
                },
                want_infos: vec![],
                want_err: Some("apply failed"),
            },
            Case {
                name: "invalid JSON",
                targets: vec![target("default", "deployment", "myapp")],
                mock_data: CopyMocksData {
                    get_json: Some("not-json".to_string()),
                    ..Default::default()
                },
                want_infos: vec![],
                want_err: Some("parsing deployment JSON for deployment/myapp"),
            },
            Case {
                name: "empty targets",
                targets: vec![],
                mock_data: CopyMocksData::default(),
                want_infos: vec![],
                want_err: None,
            },
            Case {
                name: "multiple targets",
                targets: vec![
                    target("ns1", "deployment", "app1"),
                    target("ns1", "deployment", "app2"),
                    target("ns2", "statefulset", "db1"),
                ],
                mock_data: CopyMocksData::default(),
                want_infos: vec![
                    info("deployment", "app1", "ns1"),
                    info("deployment", "app2", "ns1"),
                    info("statefulset", "db1", "ns2"),
                ],
                want_err: None,
            },
        ];

        for case in cases {
            let builder = setup_deduplicated_copy_mocks(
                MockKubectlBackend::new(),
                &case.targets,
                &case.mock_data,
            );

            let mgr = Manager::new(Box::new(builder));

            let result = mgr.prepare_copies(&case.targets, TEST_COPY_IMAGE);

            if let Some(want_err) = case.want_err {
                let err = result.unwrap_err();
                assert!(
                    err.to_string().contains(want_err),
                    "{}: expected error containing {want_err:?}, got {err}",
                    case.name
                );
                continue;
            }

            let infos = result.unwrap_or_else(|e| panic!("{}: unexpected error: {e}", case.name));
            assert_eq!(infos, case.want_infos, "{}", case.name);
        }
    }

    fn setup_cleanup_mocks(
        info: &CopyInfo,
        delete_err: bool,
        scale_err: bool,
        wait_err: bool,
    ) -> MockKubectlBackend {
        let mut builder = MockKubectlBackend::new().expect_delete(move |_rt, _name, _ns| {
            if delete_err {
                Err(kubectl_err("delete failed"))
            } else {
                Ok(())
            }
        });

        if delete_err {
            return builder;
        }

        builder = builder.expect_scale(move |_rt, _name, _ns, replicas| {
            assert_eq!(replicas, 1);
            if scale_err {
                Err(kubectl_err("scale failed"))
            } else {
                Ok(())
            }
        });

        if scale_err {
            return builder;
        }

        let _ = info;

        builder.expect_wait(move |_rt, _name, _ns, condition| {
            assert_eq!(condition, "available");
            if wait_err {
                Err(kubectl_err("wait failed"))
            } else {
                Ok(())
            }
        })
    }

    // Ported from internal/reloader/mirror/copy_test.go:443 (TestCleanupCopy)
    #[test]
    fn test_cleanup_copy() {
        let copy_info = CopyInfo {
            resource_type: "deployment".to_string(),
            name: "myapp".to_string(),
            namespace: "default".to_string(),
        };

        struct Case {
            name: &'static str,
            delete_err: bool,
            scale_err: bool,
            wait_err: bool,
            want_err: Option<&'static str>,
        }

        let cases = vec![
            Case {
                name: "success",
                delete_err: false,
                scale_err: false,
                wait_err: false,
                want_err: None,
            },
            Case {
                name: "delete error",
                delete_err: true,
                scale_err: false,
                wait_err: false,
                want_err: Some("delete failed"),
            },
            Case {
                name: "scale error",
                delete_err: false,
                scale_err: true,
                wait_err: false,
                want_err: Some("scale failed"),
            },
            Case {
                name: "wait error",
                delete_err: false,
                scale_err: false,
                wait_err: true,
                want_err: Some("wait failed"),
            },
        ];

        for case in cases {
            let mock_kubectl =
                setup_cleanup_mocks(&copy_info, case.delete_err, case.scale_err, case.wait_err);
            let mgr = Manager::new(Box::new(mock_kubectl));

            let result = mgr.cleanup_copy(&copy_info);

            if let Some(want_err) = case.want_err {
                let err = result.unwrap_err();
                assert!(
                    err.to_string().contains(want_err),
                    "{}: expected error containing {want_err:?}, got {err}",
                    case.name
                );
                continue;
            }

            result.unwrap_or_else(|e| panic!("{}: unexpected error: {e}", case.name));
        }
    }

    // Ported from internal/reloader/mirror/copy_test.go:498 (TestCollectAndValidateTargets)
    #[test]
    fn test_collect_and_validate_targets() {
        fn mirror_reloader(
            ns: &str,
            res_type: &str,
            name: &str,
            container: &str,
        ) -> ReloaderConfig {
            ReloaderConfig {
                reloader_type: "mirror".to_string(),
                mirror: Some(MirrorConfig {
                    target: Some(MirrorTargetConfig {
                        namespace: ns.to_string(),
                        kubernetes_type: res_type.to_string(),
                        name: name.to_string(),
                        container: container.to_string(),
                    }),
                    ..Default::default()
                }),
                ..Default::default()
            }
        }

        fn target_with(ns: &str, res_type: &str, name: &str, containers: &[&str]) -> Target {
            Target {
                namespace: ns.to_string(),
                kubernetes_type: res_type.to_string(),
                name: name.to_string(),
                containers: containers.iter().map(|s| s.to_string()).collect(),
                ..Default::default()
            }
        }

        struct Case {
            name: &'static str,
            reloaders: HashMap<String, ReloaderConfig>,
            want_targets: Vec<Target>,
            want_err: bool,
        }

        let cases = vec![
            Case {
                name: "two reloaders same pod different containers merged",
                reloaders: HashMap::from([
                    (
                        "reloader-a".to_string(),
                        mirror_reloader("ns", "deployment", "app", "main"),
                    ),
                    (
                        "reloader-b".to_string(),
                        mirror_reloader("ns", "deployment", "app", "sidecar"),
                    ),
                ]),
                want_targets: vec![target_with("ns", "deployment", "app", &["main", "sidecar"])],
                want_err: false,
            },
            Case {
                name: "reloaders targeting different pods produce separate targets",
                reloaders: HashMap::from([
                    (
                        "a".to_string(),
                        mirror_reloader("ns", "deployment", "svc-a", "c1"),
                    ),
                    (
                        "b".to_string(),
                        mirror_reloader("ns", "deployment", "svc-b", "c2"),
                    ),
                ]),
                want_targets: vec![
                    target_with("ns", "deployment", "svc-a", &["c1"]),
                    target_with("ns", "deployment", "svc-b", &["c2"]),
                ],
                want_err: false,
            },
            Case {
                name: "single container target",
                reloaders: HashMap::from([(
                    "a".to_string(),
                    mirror_reloader("ns", "deployment", "app", "web"),
                )]),
                want_targets: vec![target_with("ns", "deployment", "app", &["web"])],
                want_err: false,
            },
            Case {
                name: "pod-level target produces nil containers",
                reloaders: HashMap::from([(
                    "a".to_string(),
                    mirror_reloader("ns", "deployment", "app", ""),
                )]),
                want_targets: vec![target_with("ns", "deployment", "app", &[])],
                want_err: false,
            },
            Case {
                name: "non-mirror reloaders are ignored",
                reloaders: HashMap::from([
                    (
                        "helm-reloader".to_string(),
                        ReloaderConfig {
                            reloader_type: "helm".to_string(),
                            ..Default::default()
                        },
                    ),
                    (
                        "skaffold".to_string(),
                        ReloaderConfig {
                            reloader_type: "skaffold".to_string(),
                            ..Default::default()
                        },
                    ),
                ]),
                want_targets: vec![],
                want_err: false,
            },
            Case {
                name: "empty reloaders returns nil",
                reloaders: HashMap::new(),
                want_targets: vec![],
                want_err: false,
            },
            Case {
                name: "duplicate container for same pod returns error",
                reloaders: HashMap::from([
                    (
                        "a".to_string(),
                        mirror_reloader("ns", "deployment", "app", "web"),
                    ),
                    (
                        "b".to_string(),
                        mirror_reloader("ns", "deployment", "app", "web"),
                    ),
                ]),
                want_targets: vec![],
                want_err: true,
            },
            Case {
                name: "pod-level target mixed with container target returns error",
                reloaders: HashMap::from([
                    (
                        "a".to_string(),
                        mirror_reloader("ns", "deployment", "app", ""),
                    ),
                    (
                        "b".to_string(),
                        mirror_reloader("ns", "deployment", "app", "web"),
                    ),
                ]),
                want_targets: vec![],
                want_err: true,
            },
            Case {
                name: "container target mixed with pod-level target returns error",
                reloaders: HashMap::from([
                    (
                        "a".to_string(),
                        mirror_reloader("ns", "deployment", "app", "web"),
                    ),
                    (
                        "b".to_string(),
                        mirror_reloader("ns", "deployment", "app", ""),
                    ),
                ]),
                want_targets: vec![],
                want_err: true,
            },
            Case {
                name: "mirror reloader with nil mirror config is skipped",
                reloaders: HashMap::from([(
                    "a".to_string(),
                    ReloaderConfig {
                        reloader_type: "mirror".to_string(),
                        mirror: None,
                        ..Default::default()
                    },
                )]),
                want_targets: vec![],
                want_err: false,
            },
            Case {
                name: "whitespace-only container name is treated as pod-level target",
                reloaders: HashMap::from([(
                    "a".to_string(),
                    mirror_reloader("ns", "deployment", "app", "  "),
                )]),
                want_targets: vec![target_with("ns", "deployment", "app", &[])],
                want_err: false,
            },
        ];

        for case in cases {
            let result = collect_and_validate_targets(&case.reloaders);

            if case.want_err {
                assert!(result.is_err(), "{}: expected an error", case.name);
                continue;
            }

            let got = result.unwrap_or_else(|e| panic!("{}: unexpected error: {e}", case.name));
            assert_eq!(got, case.want_targets, "{}", case.name);
        }
    }
}