Repository: robustmq/robustmq (language: rust)

Issue:
Test: Optimize integration tests in the Storage directory in MQTT Broker
### What type of enhancement is this?

There are some test cases in the Storage directory src/mqtt-broker/src/storage that require access to the Placement Center. For example:
![image](https://github.com/user-attachments/assets/70e58943-c53e-4b98-8760-58a70d5d2960)

We need to move these tests to the robustmq-test integration testing directory. tests/tests/mqtt_server/storage.

And open test scripts/unit-test.sh to support testing of mqtt-broker module.
![image](https://github.com/user-attachments/assets/e6863744-856c-4440-ac02-86a6449e0450)

### What does the enhancement do?

Optimize integration tests in the Storage directory in MQTT Broker

### Implementation challenges

_No response_


Discussion / hints:
mark

Fix patch:
diff --git a/src/mqtt-broker/src/storage/cluster.rs b/src/mqtt-broker/src/storage/cluster.rs
index d4c2247dc..b6c0a5e99 100644
--- a/src/mqtt-broker/src/storage/cluster.rs
+++ b/src/mqtt-broker/src/storage/cluster.rs
@@ -185,91 +185,3 @@ impl ClusterStorage {
         vec!["cluster".to_string(), cluster_name]
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use common_base::config::broker_mqtt::{broker_mqtt_conf, init_broker_mqtt_conf_by_path};
-    use grpc_clients::pool::ClientPool;
-    use metadata_struct::mqtt::cluster::{
-        MqttClusterDynamicConfig, MqttClusterDynamicConfigProtocol,
-    };
-
-    use crate::storage::cluster::ClusterStorage;
-
-    #[tokio::test]
-    #[ignore]
-    async fn cluster_node_test() {
-        let path = format!(
-            "{}/../../config/mqtt-server.toml",
-            env!("CARGO_MANIFEST_DIR")
-        );
-        init_broker_mqtt_conf_by_path(&path);
-
-        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
-        let cluster_storage = ClusterStorage::new(client_pool);
-
-        let mut config = broker_mqtt_conf().clone();
-        config.broker_id = 1234u64;
-        cluster_storage.register_node(&config).await.unwrap();
-
-        let node_list = cluster_storage.node_list().await.unwrap();
-        let register_node_exist = node_list
-            .iter()
-            .any(|node| node.node_id == config.broker_id);
-        assert!(register_node_exist);
-
-        cluster_storage.unregister_node(&config).await.unwrap();
-
-        let node_list_after_unregister = cluster_storage.node_list().await.unwrap();
-        let unregister_node_exist = node_list_after_unregister
-            .iter()
-            .any(|node| node.node_id == config.broker_id);
-        assert!(!unregister_node_exist);
-    }
-
-    #[tokio::test]
-    #[ignore]
-    async fn cluster_config_test() {
-        let path = format!(
-            "{}/../../config/mqtt-server.toml",
-            env!("CARGO_MANIFEST_DIR")
-        );
-        init_broker_mqtt_conf_by_path(&path);
-
-        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
-        let cluster_storage = ClusterStorage::new(client_pool);
-
-        let cluster_name = "robust_test".to_string();
-        let cluster = MqttClusterDynamicConfig {
-            protocol: MqttClusterDynamicConfigProtocol {
-                topic_alias_max: 999,
-                ..Default::default()
-            },
-            ..Default::default()
-        };
-        cluster_storage
-            .set_cluster_config(cluster_name.clone(), cluster)
-            .await
-            .unwrap();
-
-        let result = cluster_storage
-            .get_cluster_config(cluster_name.clone())
-            .await
-            .unwrap()
-            .unwrap();
-        assert_eq!(result.protocol.topic_alias_max, 999);
-
-        cluster_storage
-            .delete_cluster_config(cluster_name.clone())
-            .await
-            .unwrap();
-
-        let result = cluster_storage
-            .get_cluster_config(cluster_name.clone())
-            .await
-            .unwrap();
-        assert!(result.is_none());
-    }
-}
diff --git a/src/mqtt-broker/src/storage/session.rs b/src/mqtt-broker/src/storage/session.rs
index aae67c778..d08da13fe 100644
--- a/src/mqtt-broker/src/storage/session.rs
+++ b/src/mqtt-broker/src/storage/session.rs
@@ -168,81 +168,3 @@ impl SessionStorage {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use common_base::config::broker_mqtt::init_broker_mqtt_conf_by_path;
-    use common_base::tools::now_second;
-    use grpc_clients::pool::ClientPool;
-    use metadata_struct::mqtt::session::MqttSession;
-
-    use crate::storage::session::SessionStorage;
-
-    #[tokio::test]
-    #[ignore]
-    async fn session_test() {
-        let path = format!(
-            "{}/../../config/mqtt-server.toml",
-            env!("CARGO_MANIFEST_DIR")
-        );
-        init_broker_mqtt_conf_by_path(&path);
-
-        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
-        let session_storage = SessionStorage::new(client_pool);
-        let client_id: String = "client_id_11111".to_string();
-        let session = MqttSession {
-            client_id: client_id.clone(),
-            session_expiry: 1000,
-            broker_id: Some(1),
-            reconnect_time: Some(now_second()),
-            ..Default::default()
-        };
-
-        session_storage
-            .set_session(client_id.clone(), &session)
-            .await
-            .unwrap();
-
-        let result = session_storage
-            .get_session(client_id.clone())
-            .await
-            .unwrap()
-            .unwrap();
-        assert_eq!(result.client_id, client_id);
-        assert_eq!(result.broker_id.unwrap(), 1);
-        assert!(result.connection_id.is_none());
-
-        session_storage
-            .update_session(client_id.clone(), 3, 3, now_second(), 0)
-            .await
-            .unwrap();
-
-        let result = session_storage
-            .get_session(client_id.clone())
-            .await
-            .unwrap()
-            .unwrap();
-        assert_eq!(result.broker_id.unwrap(), 3);
-        assert_eq!(result.connection_id.unwrap(), 3);
-
-        let result = session_storage.list_session().await.unwrap();
-        let prefix_len = result.len();
-        assert!(!result.is_empty());
-
-        session_storage
-            .delete_session(client_id.clone())
-            .await
-            .unwrap();
-
-        let result = session_storage
-            .get_session(client_id.clone())
-            .await
-            .unwrap();
-        assert!(result.is_none());
-
-        let result = session_storage.list_session().await.unwrap();
-        assert_eq!(result.len(), prefix_len - 1);
-    }
-}
diff --git a/src/mqtt-broker/src/storage/user.rs b/src/mqtt-broker/src/storage/user.rs
index edc11b3a2..a29f4f895 100644
--- a/src/mqtt-broker/src/storage/user.rs
+++ b/src/mqtt-broker/src/storage/user.rs
@@ -111,58 +111,3 @@ impl UserStorage {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use common_base::config::broker_mqtt::init_broker_mqtt_conf_by_path;
-    use grpc_clients::pool::ClientPool;
-
-    use crate::storage::user::UserStorage;
-
-    #[tokio::test]
-    #[ignore]
-    async fn user_test() {
-        let path = format!(
-            "{}/../../config/mqtt-server.toml",
-            env!("CARGO_MANIFEST_DIR")
-        );
-        init_broker_mqtt_conf_by_path(&path);
-
-        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
-        let user_storage = UserStorage::new(client_pool);
-        let username = "test".to_string();
-        let password = "test_password".to_string();
-        let is_superuser = true;
-        let user_info = metadata_struct::mqtt::user::MqttUser {
-            username: username.clone(),
-            password: password.clone(),
-            is_superuser,
-        };
-        user_storage.save_user(user_info).await.unwrap();
-
-        let result = user_storage
-            .get_user(username.clone())
-            .await
-            .unwrap()
-            .unwrap();
-        assert_eq!(result.username, username);
-        assert_eq!(result.password, password);
-        assert_eq!(result.is_superuser, is_superuser);
-
-        let result = user_storage.user_list().await.unwrap();
-        let prev_len = result.len();
-        assert!(!result.is_empty());
-
-        user_storage.delete_user(username.clone()).await.unwrap();
-        let result = user_storage.get_user(username.clone()).await.unwrap();
-        assert!(result.is_none());
-
-        let result = user_storage.get_user(username.clone()).await.unwrap();
-        assert!(result.is_none());
-
-        let result = user_storage.user_list().await.unwrap();
-        assert_eq!(result.len(), prev_len - 1);
-    }
-}


Test patch:
diff --git a/tests/tests/mqtt_server/storage/cluster_storage_test.rs b/tests/tests/mqtt_server/storage/cluster_storage_test.rs
new file mode 100644
index 000000000..0865a33c6
--- /dev/null
+++ b/tests/tests/mqtt_server/storage/cluster_storage_test.rs
@@ -0,0 +1,92 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_base::config::broker_mqtt::{broker_mqtt_conf, init_broker_mqtt_conf_by_path};
+    use grpc_clients::pool::ClientPool;
+    use metadata_struct::mqtt::cluster::{
+        MqttClusterDynamicConfig, MqttClusterDynamicConfigProtocol,
+    };
+    use mqtt_broker::storage::cluster::ClusterStorage;
+
+    #[tokio::test]
+    async fn cluster_node_test() {
+        let path = format!("{}/../config/mqtt-server.toml", env!("CARGO_MANIFEST_DIR"));
+        init_broker_mqtt_conf_by_path(&path);
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
+        let cluster_storage = ClusterStorage::new(client_pool);
+
+        let mut config = broker_mqtt_conf().clone();
+        config.broker_id = 1234u64;
+        cluster_storage.register_node(&config).await.unwrap();
+
+        let node_list = cluster_storage.node_list().await.unwrap();
+        let register_node_exist = node_list
+            .iter()
+            .any(|node| node.node_id == config.broker_id);
+        assert!(register_node_exist);
+
+        cluster_storage.unregister_node(&config).await.unwrap();
+
+        let node_list_after_unregister = cluster_storage.node_list().await.unwrap();
+        let unregister_node_exist = node_list_after_unregister
+            .iter()
+            .any(|node| node.node_id == config.broker_id);
+        assert!(!unregister_node_exist);
+    }
+
+    #[tokio::test]
+    async fn cluster_config_test() {
+        let path = format!("{}/../config/mqtt-server.toml", env!("CARGO_MANIFEST_DIR"));
+        init_broker_mqtt_conf_by_path(&path);
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
+        let cluster_storage = ClusterStorage::new(client_pool);
+
+        let cluster_name = "robust_test".to_string();
+        let cluster = MqttClusterDynamicConfig {
+            protocol: MqttClusterDynamicConfigProtocol {
+                topic_alias_max: 999,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        cluster_storage
+            .set_cluster_config(cluster_name.clone(), cluster)
+            .await
+            .unwrap();
+
+        let result = cluster_storage
+            .get_cluster_config(cluster_name.clone())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.protocol.topic_alias_max, 999);
+
+        cluster_storage
+            .delete_cluster_config(cluster_name.clone())
+            .await
+            .unwrap();
+
+        let result = cluster_storage
+            .get_cluster_config(cluster_name.clone())
+            .await
+            .unwrap();
+        assert!(result.is_none());
+    }
+}
diff --git a/tests/tests/mqtt_server/storage/mod.rs b/tests/tests/mqtt_server/storage/mod.rs
index 788549b19..c361b2219 100644
--- a/tests/tests/mqtt_server/storage/mod.rs
+++ b/tests/tests/mqtt_server/storage/mod.rs
@@ -12,4 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod cluster_storage_test;
+pub mod session_storage_test;
 pub mod topic_storage_test;
+pub mod user_storage_test;
diff --git a/tests/tests/mqtt_server/storage/session_storage_test.rs b/tests/tests/mqtt_server/storage/session_storage_test.rs
new file mode 100644
index 000000000..79f5a32ee
--- /dev/null
+++ b/tests/tests/mqtt_server/storage/session_storage_test.rs
@@ -0,0 +1,86 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_base::config::broker_mqtt::init_broker_mqtt_conf_by_path;
+    use common_base::tools::now_second;
+    use grpc_clients::pool::ClientPool;
+    use metadata_struct::mqtt::session::MqttSession;
+    use mqtt_broker::storage::session::SessionStorage;
+
+    #[tokio::test]
+    async fn session_test() {
+        let path = format!("{}/../config/mqtt-server.toml", env!("CARGO_MANIFEST_DIR"));
+        init_broker_mqtt_conf_by_path(&path);
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
+        let session_storage = SessionStorage::new(client_pool);
+        let client_id: String = "client_id_11111".to_string();
+        let session = MqttSession {
+            client_id: client_id.clone(),
+            session_expiry: 1000,
+            broker_id: Some(1),
+            reconnect_time: Some(now_second()),
+            ..Default::default()
+        };
+
+        session_storage
+            .set_session(client_id.clone(), &session)
+            .await
+            .unwrap();
+
+        let result = session_storage
+            .get_session(client_id.clone())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.client_id, client_id);
+        assert_eq!(result.broker_id.unwrap(), 1);
+        assert!(result.connection_id.is_none());
+
+        session_storage
+            .update_session(client_id.clone(), 3, 3, now_second(), 0)
+            .await
+            .unwrap();
+
+        let result = session_storage
+            .get_session(client_id.clone())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.broker_id.unwrap(), 3);
+        assert_eq!(result.connection_id.unwrap(), 3);
+
+        let result = session_storage.list_session().await.unwrap();
+        let prefix_len = result.len();
+        assert!(!result.is_empty());
+
+        session_storage
+            .delete_session(client_id.clone())
+            .await
+            .unwrap();
+
+        let result = session_storage
+            .get_session(client_id.clone())
+            .await
+            .unwrap();
+        assert!(result.is_none());
+
+        let result = session_storage.list_session().await.unwrap();
+        assert_eq!(result.len(), prefix_len - 1);
+    }
+}
diff --git a/tests/tests/mqtt_server/storage/user_storage_test.rs b/tests/tests/mqtt_server/storage/user_storage_test.rs
new file mode 100644
index 000000000..bb55bc630
--- /dev/null
+++ b/tests/tests/mqtt_server/storage/user_storage_test.rs
@@ -0,0 +1,63 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_base::config::broker_mqtt::init_broker_mqtt_conf_by_path;
+    use grpc_clients::pool::ClientPool;
+    use mqtt_broker::storage::user::UserStorage;
+
+    #[tokio::test]
+    async fn user_test() {
+        let path = format!("{}/../config/mqtt-server.toml", env!("CARGO_MANIFEST_DIR"));
+        init_broker_mqtt_conf_by_path(&path);
+
+        let client_pool: Arc<ClientPool> = Arc::new(ClientPool::new(10));
+        let user_storage = UserStorage::new(client_pool);
+        let username = "test".to_string();
+        let password = "test_password".to_string();
+        let is_superuser = true;
+        let user_info = metadata_struct::mqtt::user::MqttUser {
+            username: username.clone(),
+            password: password.clone(),
+            is_superuser,
+        };
+        user_storage.save_user(user_info).await.unwrap();
+
+        let result = user_storage
+            .get_user(username.clone())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.username, username);
+        assert_eq!(result.password, password);
+        assert_eq!(result.is_superuser, is_superuser);
+
+        let result = user_storage.user_list().await.unwrap();
+        let prev_len = result.len();
+        assert!(!result.is_empty());
+
+        user_storage.delete_user(username.clone()).await.unwrap();
+        let result = user_storage.get_user(username.clone()).await.unwrap();
+        assert!(result.is_none());
+
+        let result = user_storage.get_user(username.clone()).await.unwrap();
+        assert!(result.is_none());
+
+        let result = user_storage.user_list().await.unwrap();
+        assert_eq!(result.len(), prev_len - 1);
+    }
+}