//! MQTT Server deployment and management for Kubernetes
//!
//! This module provides functionality to deploy and manage MQTT brokers
//! (specifically EMQX or Mosquitto) on Kubernetes clusters.
//!
//! # Example
//!
//! ```rust,ignore
//! use herosal_kubernetes::{KubernetesManager, MqttConfigBuilder};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     let km = KubernetesManager::new("mqtt-namespace").await?;
//!
//!     let config = MqttConfigBuilder::new()
//!         .admin_credentials("admin", "secretpassword")
//!         .replicas(3)
//!         .build()?;
//!
//!     let result = km.deploy_mqtt("my-mqtt", config).await?;
//!     println!("MQTT endpoint: {}", result.mqtt_endpoint);
//!     Ok(())
//! }
//! ```

use super::{KubernetesError, KubernetesManager, KubernetesResult};
use k8s_openapi::api::apps::v1::Deployment;
use k8s_openapi::api::core::v1::Pod;
use kube::api::ListParams;
use kube::Api;
use std::collections::HashMap;
use std::time::{Duration, Instant};

/// MQTT deployment configuration
#[derive(Debug, Clone)]
pub struct MqttConfig {
    /// Broker image to use (default: emqx/emqx:5.9.2)
    pub image: String,
    /// Number of replicas (default: 1)
    pub replicas: i32,
    /// MQTT port (default: 1883)
    pub mqtt_port: i32,
    /// MQTT over WebSocket port (default: 8083)
    pub ws_port: i32,
    /// Dashboard port for EMQX (default: 18083)
    pub dashboard_port: i32,
    /// Enable TLS (default: false)
    pub tls_enabled: bool,
    /// TLS port (default: 8883)
    pub tls_port: i32,
    /// Admin username (default: admin)
    pub admin_username: String,
    /// Admin password (required)
    pub admin_password: String,
    /// Enable anonymous access (default: false)
    pub allow_anonymous: bool,
    /// Persistence enabled (default: true)
    pub persistence_enabled: bool,
    /// Persistence storage size (default: 1Gi)
    pub persistence_size: String,
    /// Additional environment variables
    pub extra_env: HashMap<String, String>,
    /// Resource limits
    pub resource_limits: Option<ResourceLimits>,
    /// Timeout in seconds for deployment to be ready (default: 120)
    pub timeout_secs: u64,
    /// If true, delete existing deployment before creating new one (default: false)
    pub reset: bool,
}

/// Resource limits for MQTT deployment
#[derive(Debug, Clone)]
pub struct ResourceLimits {
    /// CPU request (e.g., "100m")
    pub cpu_request: String,
    /// CPU limit (e.g., "500m")
    pub cpu_limit: String,
    /// Memory request (e.g., "256Mi")
    pub memory_request: String,
    /// Memory limit (e.g., "512Mi")
    pub memory_limit: String,
}

impl Default for MqttConfig {
    fn default() -> Self {
        Self {
            image: "emqx/emqx:5.9.2".to_string(),
            replicas: 1,
            mqtt_port: 1883,
            ws_port: 8083,
            dashboard_port: 18083,
            tls_enabled: false,
            tls_port: 8883,
            admin_username: "admin".to_string(),
            admin_password: String::new(),
            allow_anonymous: false,
            persistence_enabled: true,
            persistence_size: "1Gi".to_string(),
            extra_env: HashMap::new(),
            resource_limits: None,
            timeout_secs: 120,
            reset: false,
        }
    }
}

/// Builder for MQTT configuration
pub struct MqttConfigBuilder {
    config: MqttConfig,
}

impl MqttConfigBuilder {
    /// Create a new builder with default configuration
    pub fn new() -> Self {
        Self {
            config: MqttConfig::default(),
        }
    }

    /// Set the container image
    pub fn image(mut self, image: impl Into<String>) -> Self {
        self.config.image = image.into();
        self
    }

    /// Use Mosquitto instead of EMQX
    pub fn use_mosquitto(mut self) -> Self {
        self.config.image = "eclipse-mosquitto:2".to_string();
        self
    }

    /// Set the number of replicas
    pub fn replicas(mut self, replicas: i32) -> Self {
        self.config.replicas = replicas;
        self
    }

    /// Set admin credentials
    pub fn admin_credentials(
        mut self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.config.admin_username = username.into();
        self.config.admin_password = password.into();
        self
    }

    /// Set the MQTT port
    pub fn mqtt_port(mut self, port: i32) -> Self {
        self.config.mqtt_port = port;
        self
    }

    /// Set the WebSocket port
    pub fn ws_port(mut self, port: i32) -> Self {
        self.config.ws_port = port;
        self
    }

    /// Set the dashboard port (EMQX only)
    pub fn dashboard_port(mut self, port: i32) -> Self {
        self.config.dashboard_port = port;
        self
    }

    /// Enable or disable anonymous access
    pub fn allow_anonymous(mut self, allow: bool) -> Self {
        self.config.allow_anonymous = allow;
        self
    }

    /// Enable TLS
    pub fn tls_enabled(mut self, enabled: bool) -> Self {
        self.config.tls_enabled = enabled;
        self
    }

    /// Set the TLS port
    pub fn tls_port(mut self, port: i32) -> Self {
        self.config.tls_port = port;
        self
    }

    /// Configure persistence
    pub fn persistence(mut self, enabled: bool, size: impl Into<String>) -> Self {
        self.config.persistence_enabled = enabled;
        self.config.persistence_size = size.into();
        self
    }

    /// Add an extra environment variable
    pub fn extra_env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.config.extra_env.insert(key.into(), value.into());
        self
    }

    /// Set resource limits
    pub fn resource_limits(mut self, limits: ResourceLimits) -> Self {
        self.config.resource_limits = Some(limits);
        self
    }

    /// Set the timeout for deployment readiness in seconds (default: 120)
    pub fn timeout_secs(mut self, secs: u64) -> Self {
        self.config.timeout_secs = secs;
        self
    }

    /// If true, delete existing deployment before creating new one (default: false)
    pub fn reset(mut self, reset: bool) -> Self {
        self.config.reset = reset;
        self
    }

    /// Build the configuration
    pub fn build(self) -> Result<MqttConfig, &'static str> {
        if self.config.admin_password.is_empty() {
            return Err("admin_password is required");
        }
        Ok(self.config)
    }
}

impl Default for MqttConfigBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Result of MQTT deployment containing connection information
#[derive(Debug, Clone)]
pub struct MqttDeploymentResult {
    /// Name of the deployment
    pub name: String,
    /// Namespace where deployed
    pub namespace: String,
    /// Internal MQTT endpoint (service DNS)
    pub mqtt_endpoint: String,
    /// MQTT port
    pub mqtt_port: i32,
    /// Internal WebSocket endpoint
    pub ws_endpoint: String,
    /// WebSocket port
    pub ws_port: i32,
    /// Dashboard URL (if EMQX)
    pub dashboard_url: Option<String>,
    /// Dashboard port (if EMQX)
    pub dashboard_port: Option<i32>,
    /// External endpoint (if LoadBalancer assigned)
    pub external_endpoint: Option<String>,
    /// Admin username
    pub admin_username: String,
    /// Admin password
    pub admin_password: String,
    /// Whether the broker is EMQX (true) or Mosquitto (false)
    pub is_emqx: bool,
}

impl MqttDeploymentResult {
    /// Get the connection string for MQTT clients (host:port format)
    pub fn connection_string(&self) -> String {
        format!("{}:{}", self.mqtt_endpoint, self.mqtt_port)
    }

    /// Get the WebSocket connection string
    pub fn ws_connection_string(&self) -> String {
        format!("{}:{}", self.ws_endpoint, self.ws_port)
    }

    /// Get the dashboard URL if available
    pub fn get_dashboard_url(&self) -> Option<String> {
        self.dashboard_url.clone()
    }

    /// Display formatted deployment information
    pub fn display(&self) -> String {
        let mut output = String::new();
        output.push_str(&format!("MQTT Deployment: {}\n", self.name));
        output.push_str(&format!("Namespace: {}\n", self.namespace));
        output.push_str(&format!(
            "Broker Type: {}\n",
            if self.is_emqx { "EMQX" } else { "Mosquitto" }
        ));
        output.push_str(&format!(
            "MQTT Endpoint: {}:{}\n",
            self.mqtt_endpoint, self.mqtt_port
        ));
        output.push_str(&format!(
            "WebSocket Endpoint: {}:{}\n",
            self.ws_endpoint, self.ws_port
        ));
        if let Some(ref url) = self.dashboard_url {
            output.push_str(&format!("Dashboard: {}\n", url));
        }
        if let Some(ref ext) = self.external_endpoint {
            output.push_str(&format!("External Endpoint: {}\n", ext));
        }
        output.push_str(&format!("Admin Username: {}\n", self.admin_username));
        output
    }
}

impl KubernetesManager {
    /// Deploy an MQTT broker to Kubernetes
    ///
    /// This function deploys an MQTT broker (EMQX or Mosquitto) and waits for it
    /// to be ready before returning. The function will timeout if the broker
    /// doesn't become ready within the configured timeout period.
    ///
    /// # Arguments
    ///
    /// * `name` - Name for the MQTT deployment
    /// * `config` - MQTT configuration
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<MqttDeploymentResult>` - Deployment result with connection info
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use herosal_kubernetes::{KubernetesManager, MqttConfigBuilder};
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("mqtt").await?;
    ///
    ///     let config = MqttConfigBuilder::new()
    ///         .admin_credentials("admin", "password123")
    ///         .timeout_secs(120)
    ///         .build()?;
    ///
    ///     let result = km.deploy_mqtt("mqtt-broker", config).await?;
    ///     println!("Deployed MQTT broker at: {}", result.mqtt_endpoint);
    ///     Ok(())
    /// }
    /// ```
    pub async fn deploy_mqtt(
        &self,
        name: &str,
        config: MqttConfig,
    ) -> KubernetesResult<MqttDeploymentResult> {
        log::info!(
            "Deploying MQTT broker '{}' in namespace '{}' (timeout: {}s, reset: {})",
            name,
            self.namespace(),
            config.timeout_secs,
            config.reset
        );

        let is_emqx = config.image.contains("emqx");
        let timeout = Duration::from_secs(config.timeout_secs);

        // If reset flag is set, delete existing deployment first
        if config.reset {
            log::info!(
                "Reset flag set, removing existing MQTT deployment '{}'...",
                name
            );
            // Ignore errors - deployment might not exist
            let _ = self.delete_mqtt(name).await;
            // Give Kubernetes a moment to clean up
            tokio::time::sleep(Duration::from_secs(2)).await;
        }

        // Build labels
        let mut labels = HashMap::new();
        labels.insert("app".to_string(), name.to_string());
        labels.insert("component".to_string(), "mqtt-broker".to_string());

        // Build environment variables
        let env_vars = self.build_mqtt_env_vars(&config);

        // Create the deployment
        self.deployment_create(
            name,
            &config.image,
            config.replicas,
            Some(labels.clone()),
            Some(env_vars),
        )
        .await?;

        // Create ClusterIP service for internal access
        self.service_create(
            name,
            labels.clone(),
            config.mqtt_port,
            Some(config.mqtt_port),
        )
        .await?;

        // Create LoadBalancer service for external MQTT access
        self.service_create_lb(
            &format!("{}-external", name),
            labels.clone(),
            config.mqtt_port,
            Some(config.mqtt_port),
        )
        .await?;

        // Create WebSocket service if using EMQX
        if is_emqx {
            self.service_create(
                &format!("{}-ws", name),
                labels.clone(),
                config.ws_port,
                Some(config.ws_port),
            )
            .await?;

            // Create dashboard service
            self.service_create_lb(
                &format!("{}-dashboard", name),
                labels.clone(),
                config.dashboard_port,
                Some(config.dashboard_port),
            )
            .await?;
        }

        // Wait for deployment to be ready
        log::info!("Waiting for MQTT deployment '{}' to be ready...", name);
        self.wait_for_mqtt_deployment_ready(name, config.replicas, timeout)
            .await?;

        // Build the result
        let namespace = self.namespace();
        let mqtt_endpoint = format!("{}.{}.svc.cluster.local", name, namespace);
        let ws_endpoint = if is_emqx {
            format!("{}-ws.{}.svc.cluster.local", name, namespace)
        } else {
            mqtt_endpoint.clone()
        };

        let dashboard_url = if is_emqx {
            Some(format!(
                "http://{}-dashboard.{}.svc.cluster.local:{}",
                name, namespace, config.dashboard_port
            ))
        } else {
            None
        };

        // Try to get external endpoint (may take some time for LoadBalancer)
        let external_endpoint = self
            .get_mqtt_external_endpoint(&format!("{}-external", name))
            .await
            .ok();

        let result = MqttDeploymentResult {
            name: name.to_string(),
            namespace: namespace.to_string(),
            mqtt_endpoint,
            mqtt_port: config.mqtt_port,
            ws_endpoint,
            ws_port: config.ws_port,
            dashboard_url,
            dashboard_port: if is_emqx {
                Some(config.dashboard_port)
            } else {
                None
            },
            external_endpoint,
            admin_username: config.admin_username,
            admin_password: config.admin_password,
            is_emqx,
        };

        log::info!(
            "MQTT broker '{}' deployed and ready. Endpoint: {}:{}",
            name,
            result.mqtt_endpoint,
            result.mqtt_port
        );

        Ok(result)
    }

    /// Wait for MQTT deployment to be ready
    ///
    /// Polls the deployment and pod status until all replicas are ready or an error is detected.
    /// Immediately fails if pods are in an error state (ImagePullBackOff, CrashLoopBackOff, etc.)
    async fn wait_for_mqtt_deployment_ready(
        &self,
        name: &str,
        expected_replicas: i32,
        timeout: Duration,
    ) -> KubernetesResult<()> {
        let deployments: Api<Deployment> = Api::namespaced(self.client().clone(), self.namespace());
        let pods: Api<Pod> = Api::namespaced(self.client().clone(), self.namespace());
        let poll_interval = Duration::from_secs(2);
        let start = Instant::now();

        loop {
            // First, check pod status for any errors
            let label_selector = format!("app={}", name);
            let lp = ListParams::default().labels(&label_selector);

            if let Ok(pod_list) = pods.list(&lp).await {
                for pod in &pod_list.items {
                    let pod_name = pod.metadata.name.as_deref().unwrap_or("unknown");

                    if let Some(status) = &pod.status {
                        // Check container statuses for errors
                        if let Some(container_statuses) = &status.container_statuses {
                            for cs in container_statuses {
                                if let Some(waiting) =
                                    &cs.state.as_ref().and_then(|s| s.waiting.as_ref())
                                {
                                    let reason = waiting.reason.as_deref().unwrap_or("Unknown");
                                    let message = waiting.message.as_deref().unwrap_or("");

                                    // Check for fatal errors that won't resolve on their own
                                    match reason {
                                        "ImagePullBackOff" | "ErrImagePull"
                                        | "InvalidImageName" => {
                                            return Err(KubernetesError::operation_error(format!(
                                                "Pod '{}' failed to pull image: {} - {}. \
                                                 Check that the image '{}' exists and is accessible.",
                                                pod_name, reason, message,
                                                &cs.image
                                            )));
                                        }
                                        "CrashLoopBackOff" => {
                                            return Err(KubernetesError::operation_error(format!(
                                                "Pod '{}' is crash-looping: {}. \
                                                 Check logs with: kubectl logs -n {} {}",
                                                pod_name,
                                                message,
                                                self.namespace(),
                                                pod_name
                                            )));
                                        }
                                        "CreateContainerConfigError" | "CreateContainerError" => {
                                            return Err(KubernetesError::operation_error(format!(
                                                "Pod '{}' failed to create container: {} - {}",
                                                pod_name, reason, message
                                            )));
                                        }
                                        "ContainerCreating" | "PodInitializing" => {
                                            log::info!(
                                                "Pod '{}' is starting: {}",
                                                pod_name,
                                                reason
                                            );
                                        }
                                        _ => {
                                            log::debug!(
                                                "Pod '{}' waiting: {} - {}",
                                                pod_name,
                                                reason,
                                                message
                                            );
                                        }
                                    }
                                }

                                // Check for terminated containers with errors
                                if let Some(terminated) =
                                    &cs.state.as_ref().and_then(|s| s.terminated.as_ref())
                                {
                                    if terminated.exit_code != 0 {
                                        let reason =
                                            terminated.reason.as_deref().unwrap_or("Unknown");
                                        let message = terminated.message.as_deref().unwrap_or("");
                                        return Err(KubernetesError::operation_error(format!(
                                            "Pod '{}' container terminated with error (exit code {}): {} - {}. \
                                             Check logs with: kubectl logs -n {} {}",
                                            pod_name, terminated.exit_code, reason, message,
                                            self.namespace(), pod_name
                                        )));
                                    }
                                }
                            }
                        }

                        // Check init container statuses
                        if let Some(init_statuses) = &status.init_container_statuses {
                            for cs in init_statuses {
                                if let Some(waiting) =
                                    &cs.state.as_ref().and_then(|s| s.waiting.as_ref())
                                {
                                    let reason = waiting.reason.as_deref().unwrap_or("Unknown");
                                    let message = waiting.message.as_deref().unwrap_or("");

                                    if matches!(
                                        reason,
                                        "ImagePullBackOff" | "ErrImagePull" | "CrashLoopBackOff"
                                    ) {
                                        return Err(KubernetesError::operation_error(format!(
                                            "Pod '{}' init container failed: {} - {}",
                                            pod_name, reason, message
                                        )));
                                    }
                                }
                            }
                        }
                    }
                }
            }

            // Check deployment status
            match deployments.get(name).await {
                Ok(deployment) => {
                    if let Some(status) = &deployment.status {
                        let ready_replicas = status.ready_replicas.unwrap_or(0);
                        let available_replicas = status.available_replicas.unwrap_or(0);
                        let updated_replicas = status.updated_replicas.unwrap_or(0);

                        log::info!(
                            "MQTT deployment '{}': {}/{} ready, {}/{} available, {}/{} updated",
                            name,
                            ready_replicas,
                            expected_replicas,
                            available_replicas,
                            expected_replicas,
                            updated_replicas,
                            expected_replicas
                        );

                        // Check for deployment conditions that indicate failure
                        if let Some(conditions) = &status.conditions {
                            for condition in conditions {
                                if condition.type_ == "Progressing"
                                    && condition.status == "False"
                                    && condition.reason.as_deref()
                                        == Some("ProgressDeadlineExceeded")
                                {
                                    return Err(KubernetesError::operation_error(format!(
                                        "Deployment '{}' failed to progress: {}",
                                        name,
                                        condition.message.as_deref().unwrap_or("Unknown reason")
                                    )));
                                }
                            }
                        }

                        // Check if deployment is fully ready
                        if ready_replicas >= expected_replicas
                            && available_replicas >= expected_replicas
                        {
                            log::info!("MQTT deployment '{}' is ready!", name);
                            return Ok(());
                        }
                    }
                }
                Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                    log::warn!("Deployment '{}' not found yet, waiting...", name);
                }
                Err(e) => {
                    log::warn!("Error checking deployment status: {}", e);
                }
            }

            // Check timeout
            if start.elapsed() > timeout {
                return Err(KubernetesError::timeout(format!(
                    "Timeout waiting for MQTT deployment '{}' to be ready after {:?}. \
                     Check pod status with: kubectl get pods -n {} -l app={} \
                     and logs with: kubectl logs -n {} -l app={}",
                    name,
                    timeout,
                    self.namespace(),
                    name,
                    self.namespace(),
                    name
                )));
            }

            tokio::time::sleep(poll_interval).await;
        }
    }

    /// Build environment variables for MQTT deployment
    fn build_mqtt_env_vars(&self, config: &MqttConfig) -> HashMap<String, String> {
        let mut env = HashMap::new();

        if config.image.contains("emqx") {
            // EMQX environment variables
            env.insert(
                "EMQX_DASHBOARD__DEFAULT_USERNAME".to_string(),
                config.admin_username.clone(),
            );
            env.insert(
                "EMQX_DASHBOARD__DEFAULT_PASSWORD".to_string(),
                config.admin_password.clone(),
            );
            env.insert(
                "EMQX_ALLOW_ANONYMOUS".to_string(),
                config.allow_anonymous.to_string(),
            );
            env.insert(
                "EMQX_LISTENERS__TCP__DEFAULT__BIND".to_string(),
                format!("0.0.0.0:{}", config.mqtt_port),
            );
            env.insert(
                "EMQX_LISTENERS__WS__DEFAULT__BIND".to_string(),
                format!("0.0.0.0:{}", config.ws_port),
            );
            env.insert(
                "EMQX_DASHBOARD__LISTENERS__HTTP__BIND".to_string(),
                format!("0.0.0.0:{}", config.dashboard_port),
            );
        } else {
            // Mosquitto environment variables are limited; config is typically via file
            // But we can set some basic ones
            env.insert("MQTT_PORT".to_string(), config.mqtt_port.to_string());
        }

        // Add extra env vars
        for (k, v) in &config.extra_env {
            env.insert(k.clone(), v.clone());
        }

        env
    }

    /// Get the external endpoint for MQTT service
    async fn get_mqtt_external_endpoint(&self, service_name: &str) -> KubernetesResult<String> {
        let info = self.service_connection_info(service_name).await?;
        info.external_ip
            .map(|ip| format!("{}:{}", ip, info.ports.first().unwrap_or(&1883)))
            .ok_or_else(|| {
                KubernetesError::operation_error("No external IP assigned to MQTT service")
            })
    }

    /// Delete an MQTT deployment and its associated resources
    ///
    /// # Arguments
    ///
    /// * `name` - Name of the MQTT deployment to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn delete_mqtt(&self, name: &str) -> KubernetesResult<()> {
        log::info!("Deleting MQTT deployment '{}'", name);

        // Delete all associated resources
        self.deployment_delete(name).await?;
        self.service_delete(name).await?;
        self.service_delete(&format!("{}-external", name)).await?;
        self.service_delete(&format!("{}-ws", name)).await?;
        self.service_delete(&format!("{}-dashboard", name)).await?;

        log::info!("MQTT deployment '{}' deleted successfully", name);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_mqtt_config_default() {
        let config = MqttConfig::default();
        assert_eq!(config.image, "emqx/emqx:5.9.2");
        assert_eq!(config.replicas, 1);
        assert_eq!(config.mqtt_port, 1883);
        assert_eq!(config.ws_port, 8083);
        assert_eq!(config.dashboard_port, 18083);
        assert!(!config.allow_anonymous);
        assert!(config.persistence_enabled);
    }

    #[test]
    fn test_mqtt_config_builder() {
        let config = MqttConfigBuilder::new()
            .image("emqx/emqx:5.9.2")
            .replicas(3)
            .admin_credentials("admin", "password123")
            .mqtt_port(1884)
            .allow_anonymous(true)
            .build()
            .unwrap();

        assert_eq!(config.image, "emqx/emqx:5.9.2");
        assert_eq!(config.replicas, 3);
        assert_eq!(config.admin_username, "admin");
        assert_eq!(config.admin_password, "password123");
        assert_eq!(config.mqtt_port, 1884);
        assert!(config.allow_anonymous);
    }

    #[test]
    fn test_mqtt_config_builder_requires_password() {
        let result = MqttConfigBuilder::new()
            .admin_credentials("admin", "")
            .build();

        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), "admin_password is required");
    }

    #[test]
    fn test_mqtt_config_builder_mosquitto() {
        let config = MqttConfigBuilder::new()
            .use_mosquitto()
            .admin_credentials("admin", "pass")
            .build()
            .unwrap();

        assert_eq!(config.image, "eclipse-mosquitto:2");
    }

    #[test]
    fn test_mqtt_deployment_result() {
        let result = MqttDeploymentResult {
            name: "test-mqtt".to_string(),
            namespace: "default".to_string(),
            mqtt_endpoint: "test-mqtt.default.svc.cluster.local".to_string(),
            mqtt_port: 1883,
            ws_endpoint: "test-mqtt-ws.default.svc.cluster.local".to_string(),
            ws_port: 8083,
            dashboard_url: Some(
                "http://test-mqtt-dashboard.default.svc.cluster.local:18083".to_string(),
            ),
            dashboard_port: Some(18083),
            external_endpoint: None,
            admin_username: "admin".to_string(),
            admin_password: "secret".to_string(),
            is_emqx: true,
        };

        assert_eq!(
            result.connection_string(),
            "test-mqtt.default.svc.cluster.local:1883"
        );
        assert_eq!(
            result.ws_connection_string(),
            "test-mqtt-ws.default.svc.cluster.local:8083"
        );
        assert!(result.get_dashboard_url().is_some());
    }
}
