added more unit tests (#145)

Co-authored-by: Hengfei Yang <hengfei.yang@gmail.com>

Author: oasisk

src/common/http.rs

@@ -38,3 +38,30 @@ pub fn get_stream_type_from_request(
 
     Ok(stream_type)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn test_get_file_from_cache() {
+        let key = "type".to_string();
+
+        let mut map: HashMap<String, String> = HashMap::new();
+        map.insert(key.clone(), key.clone());
+
+        let resp = get_stream_type_from_request(&Query { 0: map.clone() });
+        assert!(resp.is_err());
+
+        map.insert(key.clone(), "LOGS".to_string());
+        let resp = get_stream_type_from_request(&Query { 0: map.clone() });
+        assert_eq!(resp.unwrap(), Some(StreamType::Logs));
+
+        map.insert(key.clone(), "METRICS".to_string());
+        let resp = get_stream_type_from_request(&Query { 0: map.clone() });
+        assert_eq!(resp.unwrap(), Some(StreamType::Metrics));
+
+        map.insert(key.clone(), "TRACES".to_string());
+        let resp = get_stream_type_from_request(&Query { 0: map.clone() });
+        assert_eq!(resp.unwrap(), Some(StreamType::Traces));
+    }
+}

src/handler/grpc/mod.rs

@@ -93,3 +93,23 @@ impl From<&meta::common::FileKey> for cluster_rpc::FileKey {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    #[actix_web::test]
+    async fn test_get_file_meta() {
+        let file_meta = meta::common::FileMeta {
+            min_ts: 1667978841110,
+            max_ts: 1667978845354,
+            records: 300,
+            original_size: 10,
+            compressed_size: 1,
+        };
+
+        let rpc_meta = cluster_rpc::FileMeta::from(&file_meta);
+        let resp = meta::common::FileMeta::from(&rpc_meta);
+
+        assert_eq!(file_meta, resp);
+    }
+}

src/infra/cache/file_data.rs

@@ -136,3 +136,32 @@ pub async fn download(file: &str) -> Result<Bytes, anyhow::Error> {
     };
     Ok(data)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn test_get_file_from_cache() {
+        let mut file_data = FileData::default();
+        let file_key = "files/default/logs/olympics/2022/10/03/10/6982652937134804993_1.parquet";
+        let content = Bytes::from("Some text");
+
+        let resp = file_data.set(file_key, content.clone());
+        assert!(resp.is_ok());
+
+        let resp = file_data.get(file_key);
+        assert_eq!(resp.unwrap(), content);
+
+        let resp = set(file_key, content.clone());
+        assert!(resp.is_ok());
+
+        let resp = exist(file_key);
+        assert_eq!(resp.unwrap(), true);
+
+        let resp = get(file_key);
+        assert_eq!(resp.unwrap(), content);
+
+        let resp = stats();
+        assert!(resp.0 > 0);
+    }
+}

src/infra/cache/stats.rs

@@ -111,5 +111,36 @@ mod tests {
         let stats = get_stats();
         let data = get_stream_stats_len();
         assert_eq!(data, stats.len());
+
+        let val = StreamStats {
+            doc_time_min: 1667978841102,
+            doc_time_max: 1667978845374,
+            doc_num: 5000,
+            file_num: 1,
+            storage_size: 200.00,
+            compressed_size: 3.00,
+        };
+
+        let _ = set_stream_stats("nexus", "default", "logs", val);
+        let stats = get_stream_stats("nexus", "default", "logs");
+        assert_eq!(stats, Some(val));
+
+        let file_meta = FileMeta {
+            min_ts: 1667978841110,
+            max_ts: 1667978845354,
+            records: 300,
+            original_size: 10,
+            compressed_size: 1,
+        };
+
+        let file_key = "files/nexus/logs/default/2022/10/03/10/6982652937134804993_1.parquet";
+        let _ = incr_stream_stats(file_key, file_meta);
+
+        let stats = get_stream_stats("nexus", "default", "logs");
+        assert_eq!(stats.unwrap().doc_num, 5300);
+
+        let _ = decr_stream_stats(file_key, file_meta);
+        let stats = get_stream_stats("nexus", "default", "logs");
+        assert_eq!(stats.unwrap().doc_num, 5000);
     }
 }

src/infra/db/mod.rs

@@ -73,24 +73,21 @@ mod tests {
     use super::default;
     use bytes::Bytes;
 
-    #[tokio::test]
-    #[ignore]
+    #[actix_web::test]
     async fn test_get() {
         let db = default();
         db.put("/foo/bar", Bytes::from("hello")).await.unwrap();
         let value = db.get("/foo/bar").await.unwrap();
         assert_eq!(value, Bytes::from("hello"));
     }
 
-    #[tokio::test]
-    #[ignore]
+    #[actix_web::test]
     async fn test_put() {
         let db = default();
         assert_eq!(true, db.put("/foo/bar", Bytes::from("hello")).await.is_ok());
     }
 
-    #[tokio::test]
-    #[ignore]
+    #[actix_web::test]
     async fn test_delete() {
         let db = default();
         assert_eq!(

src/service/alert_manager.rs

@@ -25,6 +25,7 @@ use crate::meta::search::Request;
 use crate::service::search as SearchService;
 use crate::service::triggers;
 
+#[cfg_attr(coverage_nightly, no_coverage)]
 pub async fn run() -> Result<(), anyhow::Error> {
     for item in TRIGGERS.iter() {
         if !item.is_ingest_time {
@@ -36,6 +37,8 @@ pub async fn run() -> Result<(), anyhow::Error> {
     }
     Ok(())
 }
+
+#[cfg_attr(coverage_nightly, no_coverage)]
 pub async fn handle_triggers(alert_name: &str, trigger: Trigger) {
     match super::db::alerts::get(&trigger.org, &trigger.stream, &trigger.alert_name).await {
         Ok(result) => {
@@ -71,6 +74,7 @@ pub async fn handle_triggers(alert_name: &str, trigger: Trigger) {
     }
 }
 
+#[cfg_attr(coverage_nightly, no_coverage)]
 pub async fn handle_trigger(alert_name: &str, alert: Alert) {
     let mut interval = time::interval(time::Duration::from_secs(
         (alert.frequency * 60).try_into().unwrap(),

src/service/db/compact/file_list.rs

@@ -29,3 +29,18 @@ pub async fn set_offset(offset: i64) -> Result<(), anyhow::Error> {
     db.put(key, offset.to_string().into()).await?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    #[actix_web::test]
+    async fn test_files() {
+        let off_set = 100;
+
+        let _ = set_offset(off_set).await;
+        let resp = get_offset().await;
+        assert_eq!(resp.unwrap(), off_set);
+    }
+}

src/service/db/compact/files.rs

@@ -56,3 +56,21 @@ pub async fn list_offset() -> Result<Vec<(String, i64)>, anyhow::Error> {
     }
     Ok(items)
 }
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    #[actix_web::test]
+    async fn test_files() {
+        let off_set = 100;
+
+        let _ = set_offset("nexus", "default", "logs".into(), off_set).await;
+        let resp = get_offset("nexus", "default", "logs".into()).await;
+        assert_eq!(resp.unwrap(), off_set);
+
+        let resp = list_offset().await;
+        assert!(resp.unwrap().len() > 0);
+    }
+}

src/service/db/file_list/local.rs

@@ -79,3 +79,27 @@ pub async fn broadcast_cache() -> Result<(), anyhow::Error> {
     }
     super::broadcast::send(&files).await
 }
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    #[actix_web::test]
+    async fn test_files() {
+        let file_key = "files/nexus/logs/default/2022/10/03/10/6982652937134804993_1.parquet";
+
+        let file_meta = FileMeta {
+            min_ts: 1667978841110,
+            max_ts: 1667978845354,
+            records: 300,
+            original_size: 10,
+            compressed_size: 1,
+        };
+
+        let resp = set(file_key, file_meta, false).await;
+        //let resp = cache().await;
+
+        assert!(resp.is_ok());
+    }
+}

src/service/search/datafusion/exec.rs

@@ -730,11 +730,67 @@ async fn register_udf(ctx: &mut SessionContext, _org_id: &str) {
 
 #[cfg(test)]
 mod test {
+    use arrow::array::Int32Array;
+    use arrow_schema::Field;
+    use datafusion::from_slice::FromSlice;
+
     use super::*;
     #[actix_web::test]
     async fn test_register_udf() {
         let mut ctx = SessionContext::new();
         let _ = register_udf(&mut ctx, "nexus").await;
         //assert!(res)
     }
+
+    #[actix_web::test]
+    async fn test_merge_write_recordbatch() {
+        // define a schema.
+        let schema = Arc::new(Schema::new(vec![Field::new("f", DataType::Int32, false)]));
+        // define data.
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_slice([1, 10, 10, 100]))],
+        )
+        .unwrap();
+
+        let batch2 = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_slice([2, 20, 20, 200]))],
+        )
+        .unwrap();
+
+        let res = merge_write_recordbatch(&[vec![batch, batch2]]).unwrap();
+
+        assert!(!res.is_empty())
+    }
+
+    #[actix_web::test]
+    async fn test_merge() {
+        // define a schema.
+        let schema = Arc::new(Schema::new(vec![Field::new("f", DataType::Int32, false)]));
+        // define data.
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_slice([1, 10, 10, 100]))],
+        )
+        .unwrap();
+
+        let batch2 = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_slice([2, 20, 20, 200]))],
+        )
+        .unwrap();
+
+        let res = merge(
+            "dummy",
+            1,
+            100,
+            "select * from tbl limit 10",
+            &vec![vec![batch, batch2]],
+        )
+        .await
+        .unwrap();
+
+        assert!(!res.is_empty())
+    }
 }

src/service/search/datafusion/regexp_udf.rs

@@ -261,4 +261,14 @@ mod tests {
         let count = result.iter().map(|batch| batch.num_rows()).sum::<usize>();
         assert_eq!(count, 2);
     }
+
+    #[test]
+    fn parse_escape() {
+        assert_eq!(is_valid_character_after_escape('0'), true);
+        assert_eq!(is_valid_character_after_escape('8'), true);
+        assert_eq!(is_valid_character_after_escape('x'), true);
+        assert_eq!(is_valid_character_after_escape('p'), true);
+        assert_eq!(is_valid_character_after_escape('d'), true);
+        assert_eq!(is_valid_character_after_escape('a'), false);
+    }
 }

src/service/search/datafusion/storage/file_list.rs

@@ -81,5 +81,8 @@ mod tests {
 
         let get_resp = get(session_id).await;
         assert!(get_resp.unwrap().len() > 0);
+
+        let resp = clear(session_id).await;
+        assert!(resp.is_ok());
     }
 }