chore: add UTs, run fmt and clipyy (#7753)

Cargo.lock

@@ -2547,6 +2547,7 @@ dependencies = [
  "sysinfo",
  "tantivy",
  "tantivy-fst",
+ "tempfile",
  "tokio",
  "tokio-util",
  "tracing",

src/common/utils/jwt.rs

@@ -26,6 +26,15 @@ use {
 #[cfg(feature = "enterprise")]
 use crate::common::meta::user::TokenValidationResponse;
 
+#[cfg(feature = "enterprise")]
+type VerifyTokenResult = Result<
+    (
+        TokenValidationResponse,
+        Option<TokenData<HashMap<String, Value>>>,
+    ),
+    anyhow::Error,
+>;
+
 /// Convert a `KeyAlgorithm` to an `Algorithm`
 ///
 /// There's no built-in conversion from `KeyAlgorithm`
@@ -59,13 +68,7 @@ pub(crate) fn verify_decode_token(
     aud: &str,
     get_decode_token: bool,
     login_flow: bool,
-) -> Result<
-    (
-        TokenValidationResponse,
-        Option<TokenData<HashMap<String, Value>>>,
-    ),
-    anyhow::Error,
-> {
+) -> VerifyTokenResult {
     use config::meta::user::UserRole;
     use infra::errors::JwtError;
 

src/config/Cargo.toml

@@ -69,6 +69,7 @@ svix-ksuid.workspace = true
 sysinfo.workspace = true
 tantivy.workspace = true
 tantivy-fst.workspace = true
+tempfile.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 tracing.workspace = true

src/config/src/meta/dashboards/reports.rs

@@ -26,7 +26,7 @@ pub enum ReportDestination {
     Email(String), // Supports email only
 }
 
-#[derive(Serialize, Debug, Default, Deserialize, Clone, ToSchema)]
+#[derive(Serialize, Debug, Default, Deserialize, Clone, ToSchema, PartialEq)]
 pub enum ReportMediaType {
     #[default]
     #[serde(rename = "pdf")]
@@ -54,11 +54,10 @@ pub struct ReportDashboard {
 }
 
 #[derive(Serialize, Debug, Default, Deserialize, Clone, ToSchema, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
 pub enum ReportTimerangeType {
     #[default]
-    #[serde(rename = "relative")]
     Relative,
-    #[serde(rename = "absolute")]
     Absolute,
 }
 
@@ -83,19 +82,14 @@ impl Default for ReportTimerange {
 }
 
 #[derive(Serialize, Debug, Default, Deserialize, PartialEq, Clone, ToSchema)]
+#[serde(rename_all = "lowercase")]
 pub enum ReportFrequencyType {
-    #[serde(rename = "once")]
     Once,
-    #[serde(rename = "hours")]
     Hours,
-    #[serde(rename = "days")]
     Days,
-    #[serde(rename = "weeks")]
     #[default]
     Weeks,
-    #[serde(rename = "months")]
     Months,
-    #[serde(rename = "cron")]
     Cron,
 }
 
@@ -288,7 +282,7 @@ pub struct ReportListFilters {
 }
 
 impl ReportListFilters {
-    pub fn into_parmas(self, org_id: &str) -> ListReportsParams {
+    pub fn into_params(self, org_id: &str) -> ListReportsParams {
         ListReportsParams {
             org_id: org_id.to_string(),
             folder_snowflake_id: self.folder,
@@ -319,4 +313,404 @@ mod tests {
             serde_json::from_str(&json_using_alias).unwrap();
         assert_eq!(email_details, email_details_from_alias);
     }
+
+    #[test]
+    fn test_report_destination_serialization() {
+        let destination = ReportDestination::Email("test@example.com".to_string());
+        let serialized = serde_json::to_string(&destination).unwrap();
+        assert_eq!(serialized, r#"{"email":"test@example.com"}"#);
+
+        let deserialized: ReportDestination = serde_json::from_str(&serialized).unwrap();
+        assert!(matches!(deserialized, ReportDestination::Email(_)));
+    }
+
+    #[test]
+    fn test_report_media_type_default() {
+        assert_eq!(ReportMediaType::default(), ReportMediaType::Pdf);
+    }
+
+    #[test]
+    fn test_report_dashboard_variable_default() {
+        let variable = ReportDashboardVariable {
+            key: "test_key".to_string(),
+            value: "test_value".to_string(),
+            id: Some("test_id".to_string()),
+        };
+
+        assert_eq!(variable.key, "test_key");
+        assert_eq!(variable.value, "test_value");
+        assert_eq!(variable.id, Some("test_id".to_string()));
+    }
+
+    #[test]
+    fn test_report_dashboard_serialization() {
+        let dashboard = ReportDashboard {
+            dashboard: "test_dashboard".to_string(),
+            folder: "test_folder".to_string(),
+            tabs: vec!["tab1".to_string(), "tab2".to_string()],
+            variables: vec![ReportDashboardVariable {
+                key: "var1".to_string(),
+                value: "value1".to_string(),
+                id: None,
+            }],
+            timerange: ReportTimerange::default(),
+        };
+
+        let serialized = serde_json::to_string(&dashboard).unwrap();
+        let deserialized: ReportDashboard = serde_json::from_str(&serialized).unwrap();
+
+        assert_eq!(deserialized.dashboard, "test_dashboard");
+        assert_eq!(deserialized.folder, "test_folder");
+        assert_eq!(deserialized.tabs.len(), 2);
+        assert_eq!(deserialized.variables.len(), 1);
+    }
+
+    #[test]
+    fn test_report_timerange_default() {
+        let timerange = ReportTimerange::default();
+        assert_eq!(timerange.range_type, ReportTimerangeType::Relative);
+        assert_eq!(timerange.period, "1w");
+        assert_eq!(timerange.from, 0);
+        assert_eq!(timerange.to, 0);
+    }
+
+    #[test]
+    fn test_report_timerange_absolute() {
+        let timerange = ReportTimerange {
+            range_type: ReportTimerangeType::Absolute,
+            period: "".to_string(),
+            from: 1640995200000000, // 2022-01-01 00:00:00 UTC in microseconds
+            to: 1641081600000000,   // 2022-01-02 00:00:00 UTC in microseconds
+        };
+
+        assert_eq!(timerange.range_type, ReportTimerangeType::Absolute);
+        assert_eq!(timerange.from, 1640995200000000);
+        assert_eq!(timerange.to, 1641081600000000);
+    }
+
+    #[test]
+    fn test_report_timerange_type_default() {
+        assert_eq!(
+            ReportTimerangeType::default(),
+            ReportTimerangeType::Relative
+        );
+    }
+
+    #[test]
+    fn test_report_frequency_type_default() {
+        assert_eq!(ReportFrequencyType::default(), ReportFrequencyType::Weeks);
+    }
+
+    #[test]
+    fn test_report_frequency_default() {
+        let frequency = ReportFrequency::default();
+        assert_eq!(frequency.interval, 1);
+        assert_eq!(frequency.cron, "");
+        assert_eq!(frequency.frequency_type, ReportFrequencyType::Weeks);
+        assert_eq!(frequency.align_time, default_align_time());
+    }
+
+    #[test]
+    fn test_report_frequency_cron() {
+        let frequency = ReportFrequency {
+            interval: 0,
+            cron: "0 0 * * *".to_string(),
+            frequency_type: ReportFrequencyType::Cron,
+            align_time: false,
+        };
+
+        assert_eq!(frequency.cron, "0 0 * * *");
+        assert_eq!(frequency.frequency_type, ReportFrequencyType::Cron);
+        assert!(!frequency.align_time);
+    }
+
+    #[test]
+    fn test_report_default() {
+        let report = Report::default();
+        assert!(report.name.is_empty());
+        assert!(report.title.is_empty());
+        assert!(report.org_id.is_empty());
+        assert!(report.destinations.is_empty());
+        assert!(report.dashboards.is_empty());
+        assert!(report.description.is_empty());
+        assert!(report.message.is_empty());
+        assert!(!report.enabled);
+        assert_eq!(report.media_type, ReportMediaType::Pdf);
+        assert!(report.timezone.is_empty());
+        assert_eq!(report.tz_offset, 0);
+        assert!(report.updated_at.is_none());
+        assert!(report.owner.is_empty());
+        assert!(report.last_edited_by.is_empty());
+    }
+
+    #[test]
+    fn test_report_with_data() {
+        let now = Utc::now().timestamp_micros();
+        let report = Report {
+            name: "test_report".to_string(),
+            title: "Test Report".to_string(),
+            org_id: "test_org".to_string(),
+            frequency: ReportFrequency::default(),
+            start: now,
+            dashboards: vec![ReportDashboard {
+                dashboard: "dashboard1".to_string(),
+                folder: "folder1".to_string(),
+                tabs: vec!["tab1".to_string()],
+                variables: vec![],
+                timerange: ReportTimerange::default(),
+            }],
+            destinations: vec![ReportDestination::Email("test@example.com".to_string())],
+            description: "Test description".to_string(),
+            message: "Test message".to_string(),
+            enabled: true,
+            media_type: ReportMediaType::Pdf,
+            timezone: "UTC".to_string(),
+            tz_offset: 0,
+            created_at: datetime_now(),
+            updated_at: None,
+            owner: "test_owner".to_string(),
+            last_edited_by: "test_editor".to_string(),
+        };
+
+        assert_eq!(report.name, "test_report");
+        assert_eq!(report.title, "Test Report");
+        assert_eq!(report.org_id, "test_org");
+        assert_eq!(report.start, now);
+        assert_eq!(report.dashboards.len(), 1);
+        assert_eq!(report.destinations.len(), 1);
+        assert_eq!(report.description, "Test description");
+        assert_eq!(report.message, "Test message");
+        assert!(report.enabled);
+        assert_eq!(report.timezone, "UTC");
+        assert_eq!(report.tz_offset, 0);
+        assert_eq!(report.owner, "test_owner");
+        assert_eq!(report.last_edited_by, "test_editor");
+    }
+
+    #[test]
+    fn test_report_serialization() {
+        let report = Report {
+            name: "test_report".to_string(),
+            title: "Test Report".to_string(),
+            org_id: "test_org".to_string(),
+            frequency: ReportFrequency::default(),
+            start: Utc::now().timestamp_micros(),
+            dashboards: vec![],
+            destinations: vec![ReportDestination::Email("test@example.com".to_string())],
+            description: "Test description".to_string(),
+            message: "Test message".to_string(),
+            enabled: true,
+            media_type: ReportMediaType::Pdf,
+            timezone: "UTC".to_string(),
+            tz_offset: 0,
+            created_at: datetime_now(),
+            updated_at: None,
+            owner: "test_owner".to_string(),
+            last_edited_by: "test_editor".to_string(),
+        };
+
+        let serialized = serde_json::to_string(&report).unwrap();
+        let deserialized: Report = serde_json::from_str(&serialized).unwrap();
+
+        assert_eq!(deserialized.name, "test_report");
+        assert_eq!(deserialized.title, "Test Report");
+        assert_eq!(deserialized.org_id, "test_org");
+        assert!(deserialized.enabled);
+    }
+
+    #[test]
+    fn test_report_email_details() {
+        let email_details = ReportEmailDetails {
+            recipients: vec![
+                "user1@example.com".to_string(),
+                "user2@example.com".to_string(),
+            ],
+            title: "Test Email".to_string(),
+            name: "test_name".to_string(),
+            message: "Test message".to_string(),
+            dashb_url: "https://example.com/dashboard".to_string(),
+        };
+
+        assert_eq!(email_details.recipients.len(), 2);
+        assert_eq!(email_details.title, "Test Email");
+        assert_eq!(email_details.name, "test_name");
+        assert_eq!(email_details.message, "Test message");
+        assert_eq!(email_details.dashb_url, "https://example.com/dashboard");
+    }
+
+    #[test]
+    fn test_http_report_payload() {
+        let email_details = ReportEmailDetails {
+            recipients: vec!["test@example.com".to_string()],
+            title: "Test".to_string(),
+            name: "test".to_string(),
+            message: "test".to_string(),
+            dashb_url: "https://example.com".to_string(),
+        };
+
+        let payload = HttpReportPayload {
+            dashboards: vec![ReportDashboard {
+                dashboard: "dashboard1".to_string(),
+                folder: "folder1".to_string(),
+                tabs: vec!["tab1".to_string()],
+                variables: vec![],
+                timerange: ReportTimerange::default(),
+            }],
+            email_details: email_details.clone(),
+        };
+
+        assert_eq!(payload.dashboards.len(), 1);
+        assert_eq!(payload.email_details.recipients, email_details.recipients);
+    }
+
+    #[test]
+    fn test_list_reports_params_new() {
+        let params = ListReportsParams::new("test_org");
+        assert_eq!(params.org_id, "test_org");
+        assert!(params.folder_snowflake_id.is_none());
+        assert!(params.dashboard_snowflake_id.is_none());
+        assert!(params.page_size_and_idx.is_none());
+        assert!(params.has_destinations.is_none());
+    }
+
+    #[test]
+    fn test_list_reports_params_builder_methods() {
+        let params = ListReportsParams::new("test_org")
+            .in_folder("folder_123")
+            .for_dashboard("dashboard_456")
+            .has_destinations(true)
+            .paginate(10, 0);
+
+        assert_eq!(params.org_id, "test_org");
+        assert_eq!(params.folder_snowflake_id, Some("folder_123".to_string()));
+        assert_eq!(
+            params.dashboard_snowflake_id,
+            Some("dashboard_456".to_string())
+        );
+        assert_eq!(params.has_destinations, Some(true));
+        assert_eq!(params.page_size_and_idx, Some((10, 0)));
+    }
+
+    #[test]
+    fn test_list_report_item() {
+        let item = ListReportItem {
+            name: "test_report".to_string(),
+            owner: "test_owner".to_string(),
+            description: Some("Test description".to_string()),
+            last_triggered_at: 1640995200000000,
+        };
+
+        assert_eq!(item.name, "test_report");
+        assert_eq!(item.owner, "test_owner");
+        assert_eq!(item.description, Some("Test description".to_string()));
+        assert_eq!(item.last_triggered_at, 1640995200000000);
+    }
+
+    #[test]
+    fn test_report_list_filters() {
+        let filters = ReportListFilters {
+            dashboard: Some("dashboard_123".to_string()),
+            folder: Some("folder_456".to_string()),
+            destination_less: Some(true),
+        };
+
+        let params = filters.into_params("test_org");
+        assert_eq!(params.org_id, "test_org");
+        assert_eq!(
+            params.dashboard_snowflake_id,
+            Some("dashboard_123".to_string())
+        );
+        assert_eq!(params.folder_snowflake_id, Some("folder_456".to_string()));
+        assert_eq!(params.has_destinations, Some(false)); // destination_less = true means has_destinations = false
+    }
+
+    #[test]
+    fn test_report_timerange_equality() {
+        let timerange1 = ReportTimerange {
+            range_type: ReportTimerangeType::Relative,
+            period: "1w".to_string(),
+            from: 0,
+            to: 0,
+        };
+
+        let timerange2 = ReportTimerange {
+            range_type: ReportTimerangeType::Relative,
+            period: "1w".to_string(),
+            from: 0,
+            to: 0,
+        };
+
+        assert_eq!(timerange1, timerange2);
+    }
+
+    #[test]
+    fn test_report_frequency_type_equality() {
+        assert_eq!(ReportFrequencyType::Once, ReportFrequencyType::Once);
+        assert_ne!(ReportFrequencyType::Once, ReportFrequencyType::Weeks);
+        assert_ne!(ReportFrequencyType::Hours, ReportFrequencyType::Days);
+    }
+
+    #[test]
+    fn test_report_media_type_clone() {
+        let media_type = ReportMediaType::Pdf;
+        let cloned = media_type.clone();
+        assert_eq!(media_type, cloned);
+    }
+
+    #[test]
+    fn test_report_dashboard_variable_equality() {
+        let var1 = ReportDashboardVariable {
+            key: "test_key".to_string(),
+            value: "test_value".to_string(),
+            id: Some("test_id".to_string()),
+        };
+
+        let var2 = ReportDashboardVariable {
+            key: "test_key".to_string(),
+            value: "test_value".to_string(),
+            id: Some("test_id".to_string()),
+        };
+
+        assert_eq!(var1, var2);
+    }
+
+    #[test]
+    fn test_report_dashboard_variable_without_id() {
+        let variable = ReportDashboardVariable {
+            key: "test_key".to_string(),
+            value: "test_value".to_string(),
+            id: None,
+        };
+
+        assert_eq!(variable.key, "test_key");
+        assert_eq!(variable.value, "test_value");
+        assert!(variable.id.is_none());
+    }
+
+    #[test]
+    fn test_report_timerange_type_serialization() {
+        let relative = ReportTimerangeType::Relative;
+        let absolute = ReportTimerangeType::Absolute;
+
+        let relative_json = serde_json::to_string(&relative).unwrap();
+        let absolute_json = serde_json::to_string(&absolute).unwrap();
+
+        assert_eq!(relative_json, r#""relative""#);
+        assert_eq!(absolute_json, r#""absolute""#);
+    }
+
+    #[test]
+    fn test_report_frequency_type_serialization() {
+        let once = ReportFrequencyType::Once;
+        let weeks = ReportFrequencyType::Weeks;
+        let cron = ReportFrequencyType::Cron;
+
+        let once_json = serde_json::to_string(&once).unwrap();
+        let weeks_json = serde_json::to_string(&weeks).unwrap();
+        let cron_json = serde_json::to_string(&cron).unwrap();
+
+        assert_eq!(once_json, r#""once""#);
+        assert_eq!(weeks_json, r#""weeks""#);
+        assert_eq!(cron_json, r#""cron""#);
+    }
 }

src/config/src/meta/destinations.rs

@@ -110,31 +110,28 @@ pub struct AwsSns {
 }
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "lowercase")]
 pub enum HTTPType {
     #[default]
-    #[serde(rename = "post")]
     POST,
-    #[serde(rename = "put")]
     PUT,
-    #[serde(rename = "get")]
     GET,
 }
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "lowercase")]
 pub enum HTTPOutputFormat {
     #[default]
-    #[serde(rename = "json")]
     JSON,
-    #[serde(rename = "ndjson")]
     NDJSON,
 }
 
 impl fmt::Display for HTTPType {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         match self {
-            HTTPType::POST => write!(f, "post"),
-            HTTPType::PUT => write!(f, "put"),
-            HTTPType::GET => write!(f, "get"),
+            Self::POST => write!(f, "post"),
+            Self::PUT => write!(f, "put"),
+            Self::GET => write!(f, "get"),
         }
     }
 }
@@ -153,13 +150,13 @@ pub struct Template {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, Default)]
+#[serde(rename_all = "lowercase")]
 pub enum TemplateType {
     #[default]
-    #[serde(rename = "http")]
     Http,
-    #[serde(rename = "email")]
-    Email { title: String },
-    #[serde(rename = "sns")]
+    Email {
+        title: String,
+    },
     Sns,
 }
 
@@ -172,3 +169,350 @@ impl fmt::Display for TemplateType {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use svix_ksuid::KsuidLike;
+
+    use super::*;
+
+    #[test]
+    fn test_destination_default() {
+        let dest = Destination::default();
+        assert!(dest.id.is_none());
+        assert!(dest.org_id.is_empty());
+        assert!(dest.name.is_empty());
+        assert!(matches!(dest.module, Module::Alert { .. }));
+    }
+
+    #[test]
+    fn test_destination_is_alert_destinations() {
+        let alert_dest = Destination {
+            id: None,
+            org_id: "test_org".to_string(),
+            name: "alert_dest".to_string(),
+            module: Module::Alert {
+                template: "template".to_string(),
+                destination_type: DestinationType::Http(Endpoint::default()),
+            },
+        };
+        assert!(alert_dest.is_alert_destinations());
+
+        let pipeline_dest = Destination {
+            id: None,
+            org_id: "test_org".to_string(),
+            name: "pipeline_dest".to_string(),
+            module: Module::Pipeline {
+                endpoint: Endpoint::default(),
+            },
+        };
+        assert!(!pipeline_dest.is_alert_destinations());
+    }
+
+    #[test]
+    fn test_module_default() {
+        let module = Module::default();
+        match module {
+            Module::Alert {
+                template,
+                destination_type,
+            } => {
+                assert_eq!(template, "");
+                assert!(matches!(destination_type, DestinationType::Http(_)));
+            }
+            _ => panic!("Default should be Alert"),
+        }
+    }
+
+    #[test]
+    fn test_module_display() {
+        let alert_module = Module::Alert {
+            template: "test".to_string(),
+            destination_type: DestinationType::Http(Endpoint::default()),
+        };
+        assert_eq!(alert_module.to_string(), "alert");
+
+        let pipeline_module = Module::Pipeline {
+            endpoint: Endpoint::default(),
+        };
+        assert_eq!(pipeline_module.to_string(), "pipeline");
+    }
+
+    #[test]
+    fn test_destination_type_default() {
+        let dest_type = DestinationType::default();
+        assert!(matches!(dest_type, DestinationType::Http(_)));
+    }
+
+    #[test]
+    fn test_email_default() {
+        let email = Email::default();
+        assert!(email.recipients.is_empty());
+    }
+
+    #[test]
+    fn test_endpoint_default() {
+        let endpoint = Endpoint::default();
+        assert!(endpoint.url.is_empty());
+        assert_eq!(endpoint.method, HTTPType::POST);
+        assert!(!endpoint.skip_tls_verify);
+        assert!(endpoint.headers.is_none());
+        assert!(endpoint.action_id.is_none());
+        assert!(endpoint.output_format.is_none());
+    }
+
+    #[test]
+    fn test_endpoint_with_headers() {
+        let mut headers = HashMap::new();
+        headers.insert("Authorization".to_string(), "Bearer token".to_string());
+        headers.insert("Content-Type".to_string(), "application/json".to_string());
+
+        let endpoint = Endpoint {
+            url: "https://api.example.com".to_string(),
+            method: HTTPType::POST,
+            skip_tls_verify: false,
+            headers: Some(headers.clone()),
+            action_id: Some("action_123".to_string()),
+            output_format: Some(HTTPOutputFormat::JSON),
+        };
+
+        assert_eq!(endpoint.url, "https://api.example.com");
+        assert_eq!(endpoint.method, HTTPType::POST);
+        assert!(!endpoint.skip_tls_verify);
+        assert_eq!(endpoint.headers, Some(headers));
+        assert_eq!(endpoint.action_id, Some("action_123".to_string()));
+        assert_eq!(endpoint.output_format, Some(HTTPOutputFormat::JSON));
+    }
+
+    #[test]
+    fn test_aws_sns_default() {
+        let sns = AwsSns::default();
+        assert!(sns.sns_topic_arn.is_empty());
+        assert!(sns.aws_region.is_empty());
+    }
+
+    #[test]
+    fn test_http_type_default() {
+        assert_eq!(HTTPType::default(), HTTPType::POST);
+    }
+
+    #[test]
+    fn test_http_type_display() {
+        assert_eq!(HTTPType::POST.to_string(), "post");
+        assert_eq!(HTTPType::PUT.to_string(), "put");
+        assert_eq!(HTTPType::GET.to_string(), "get");
+    }
+
+    #[test]
+    fn test_http_type_equality() {
+        assert_eq!(HTTPType::POST, HTTPType::POST);
+        assert_ne!(HTTPType::POST, HTTPType::PUT);
+        assert_ne!(HTTPType::GET, HTTPType::POST);
+    }
+
+    #[test]
+    fn test_http_output_format_default() {
+        assert_eq!(HTTPOutputFormat::default(), HTTPOutputFormat::JSON);
+    }
+
+    #[test]
+    fn test_http_output_format_equality() {
+        assert_eq!(HTTPOutputFormat::JSON, HTTPOutputFormat::JSON);
+        assert_ne!(HTTPOutputFormat::JSON, HTTPOutputFormat::NDJSON);
+    }
+
+    #[test]
+    fn test_template_default() {
+        let template = Template::default();
+        assert!(template.id.is_none());
+        assert!(template.org_id.is_empty());
+        assert!(template.name.is_empty());
+        assert!(!template.is_default);
+        assert!(matches!(template.template_type, TemplateType::Http));
+        assert!(template.body.is_empty());
+    }
+
+    #[test]
+    fn test_template_with_id() {
+        let id = svix_ksuid::Ksuid::new(None, None);
+        let template = Template {
+            id: Some(id),
+            org_id: "test_org".to_string(),
+            name: "test_template".to_string(),
+            is_default: true,
+            template_type: TemplateType::Email {
+                title: "Test Email".to_string(),
+            },
+            body: "Hello {{name}}!".to_string(),
+        };
+
+        assert_eq!(template.id, Some(id));
+        assert_eq!(template.org_id, "test_org");
+        assert_eq!(template.name, "test_template");
+        assert!(template.is_default);
+        assert!(matches!(template.template_type, TemplateType::Email { .. }));
+        assert_eq!(template.body, "Hello {{name}}!");
+    }
+
+    #[test]
+    fn test_template_type_default() {
+        assert!(matches!(TemplateType::default(), TemplateType::Http));
+    }
+
+    #[test]
+    fn test_template_type_display() {
+        assert_eq!(TemplateType::Http.to_string(), "http");
+        assert_eq!(
+            TemplateType::Email {
+                title: "Test".to_string()
+            }
+            .to_string(),
+            "email"
+        );
+        assert_eq!(TemplateType::Sns.to_string(), "sns");
+    }
+
+    #[test]
+    fn test_serialization_deserialization() {
+        let destination = Destination {
+            id: Some(svix_ksuid::Ksuid::new(None, None)),
+            org_id: "test_org".to_string(),
+            name: "test_dest".to_string(),
+            module: Module::Alert {
+                template: "alert_template".to_string(),
+                destination_type: DestinationType::Email(Email {
+                    recipients: vec!["user@example.com".to_string()],
+                }),
+            },
+        };
+
+        let serialized = serde_json::to_string(&destination).unwrap();
+        let deserialized: Destination = serde_json::from_str(&serialized).unwrap();
+
+        assert_eq!(destination.org_id, deserialized.org_id);
+        assert_eq!(destination.name, deserialized.name);
+        assert!(matches!(deserialized.module, Module::Alert { .. }));
+    }
+
+    #[test]
+    fn test_destination_type_http() {
+        let endpoint = Endpoint {
+            url: "https://webhook.example.com".to_string(),
+            method: HTTPType::POST,
+            skip_tls_verify: true,
+            headers: None,
+            action_id: None,
+            output_format: None,
+        };
+
+        let dest_type = DestinationType::Http(endpoint.clone());
+        assert!(matches!(dest_type, DestinationType::Http(_)));
+    }
+
+    #[test]
+    fn test_destination_type_email() {
+        let email = Email {
+            recipients: vec![
+                "admin@example.com".to_string(),
+                "user@example.com".to_string(),
+            ],
+        };
+
+        let dest_type = DestinationType::Email(email.clone());
+        assert!(matches!(dest_type, DestinationType::Email(_)));
+    }
+
+    #[test]
+    fn test_destination_type_sns() {
+        let sns = AwsSns {
+            sns_topic_arn: "arn:aws:sns:us-east-1:123456789012:my-topic".to_string(),
+            aws_region: "us-east-1".to_string(),
+        };
+
+        let dest_type = DestinationType::Sns(sns.clone());
+        assert!(matches!(dest_type, DestinationType::Sns(_)));
+    }
+
+    #[test]
+    fn test_module_alert() {
+        let module = Module::Alert {
+            template: "alert_template".to_string(),
+            destination_type: DestinationType::Http(Endpoint::default()),
+        };
+
+        match module {
+            Module::Alert {
+                template,
+                destination_type,
+            } => {
+                assert_eq!(template, "alert_template");
+                assert!(matches!(destination_type, DestinationType::Http(_)));
+            }
+            _ => panic!("Should be Alert module"),
+        }
+    }
+
+    #[test]
+    fn test_module_pipeline() {
+        let endpoint = Endpoint {
+            url: "https://pipeline.example.com".to_string(),
+            method: HTTPType::PUT,
+            skip_tls_verify: false,
+            headers: None,
+            action_id: None,
+            output_format: None,
+        };
+
+        let module = Module::Pipeline {
+            endpoint: endpoint.clone(),
+        };
+
+        match module {
+            Module::Pipeline {
+                endpoint: pipeline_endpoint,
+            } => {
+                assert_eq!(pipeline_endpoint.url, "https://pipeline.example.com");
+                assert_eq!(pipeline_endpoint.method, HTTPType::PUT);
+            }
+            _ => panic!("Should be Pipeline module"),
+        }
+    }
+
+    #[test]
+    fn test_template_type_email_with_title() {
+        let template_type = TemplateType::Email {
+            title: "Welcome Email".to_string(),
+        };
+
+        match template_type {
+            TemplateType::Email { title } => {
+                assert_eq!(title, "Welcome Email");
+            }
+            _ => panic!("Should be Email template type"),
+        }
+    }
+
+    #[test]
+    fn test_clone_implementations() {
+        let destination = Destination {
+            id: Some(svix_ksuid::Ksuid::new(None, None)),
+            org_id: "test_org".to_string(),
+            name: "test_dest".to_string(),
+            module: Module::Alert {
+                template: "template".to_string(),
+                destination_type: DestinationType::Http(Endpoint::default()),
+            },
+        };
+
+        let cloned = destination.clone();
+        assert_eq!(destination.org_id, cloned.org_id);
+        assert_eq!(destination.name, cloned.name);
+    }
+
+    #[test]
+    fn test_debug_implementations() {
+        let endpoint = Endpoint::default();
+        let debug_output = format!("{:?}", endpoint);
+        assert!(debug_output.contains("Endpoint"));
+    }
+}

src/config/src/meta/pipeline/mod.rs

@@ -42,8 +42,7 @@ pub type PipelineExecDFS = (
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, ToSchema)]
 pub struct Pipeline {
-    #[serde(default)]
-    #[serde(rename = "pipeline_id")]
+    #[serde(rename = "pipeline_id", default)]
     pub id: String,
     #[serde(default)]
     pub version: i32,
@@ -427,7 +426,10 @@ pub fn default_status() -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::utils::json;
+    use crate::{
+        meta::alerts::{QueryCondition, TriggerCondition},
+        utils::json,
+    };
 
     #[test]
     fn test_stream_pipeline_serialization() {
@@ -571,4 +573,865 @@ mod tests {
         let new_nodes = json::from_str::<Option<Vec<Node>>>(&nodes);
         assert!(new_nodes.is_ok());
     }
+
+    #[test]
+    fn test_pipeline_get_cache_key() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![],
+            edges: vec![],
+        };
+
+        let cache_key = pipeline.get_cache_key();
+        assert_eq!(cache_key, "test_org/test_org/test_stream/logs");
+
+        // Test scheduled pipeline
+        let derived_stream = DerivedStream {
+            org_id: "test_org".to_string(),
+            stream_type: StreamType::Logs,
+            query_condition: QueryCondition::default(),
+            trigger_condition: TriggerCondition::default(),
+            tz_offset: 0,
+            delay: None,
+            start_at: None,
+        };
+        pipeline.source = PipelineSource::Scheduled(derived_stream);
+        let cache_key = pipeline.get_cache_key();
+        assert_eq!(cache_key, "test_org/test_pipeline");
+    }
+
+    #[test]
+    fn test_pipeline_get_derived_stream() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![],
+            edges: vec![],
+        };
+
+        // Should return None for realtime pipeline
+        assert!(pipeline.get_derived_stream().is_none());
+
+        // Should return Some for scheduled pipeline
+        let derived_stream = DerivedStream {
+            org_id: "test_org".to_string(),
+            stream_type: StreamType::Logs,
+            query_condition: QueryCondition::default(),
+            trigger_condition: TriggerCondition::default(),
+            tz_offset: 0,
+            delay: None,
+            start_at: None,
+        };
+        pipeline.source = PipelineSource::Scheduled(derived_stream.clone());
+        assert_eq!(pipeline.get_derived_stream(), Some(derived_stream));
+    }
+
+    #[test]
+    fn test_pipeline_validation_empty_name() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "".to_string(), // Empty name
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![],
+            edges: vec![],
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("Please provide non-empty name")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_validation_empty_nodes() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![], // Empty nodes
+            edges: vec![],
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(result.unwrap_err().to_string().contains("Empty pipeline"));
+    }
+
+    #[test]
+    fn test_pipeline_validation_empty_edges() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream",
+                        StreamType::Logs,
+                    )),
+                    300.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+            ],
+            edges: vec![], // Empty edges
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("Please connect all Nodes")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_validation_insufficient_edges() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream",
+                        StreamType::Logs,
+                    )),
+                    300.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+                Node::new(
+                    "3".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream2",
+                        StreamType::Logs,
+                    )),
+                    500.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+            ],
+            edges: vec![
+                Edge {
+                    id: "e1-2".to_string(),
+                    source: "1".to_string(),
+                    target: "2".to_string(),
+                },
+                // Missing edge to node 3
+            ],
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("Insufficient number of edges")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_get_node_map() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream",
+                        StreamType::Logs,
+                    )),
+                    300.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+            ],
+            edges: vec![],
+        };
+
+        let node_map = pipeline.get_node_map();
+        assert_eq!(node_map.len(), 2);
+        assert!(node_map.contains_key("1"));
+        assert!(node_map.contains_key("2"));
+    }
+
+    #[test]
+    fn test_pipeline_build_adjacency_list() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream",
+                        StreamType::Logs,
+                    )),
+                    300.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+            ],
+            edges: vec![Edge {
+                id: "e1-2".to_string(),
+                source: "1".to_string(),
+                target: "2".to_string(),
+            }],
+        };
+
+        let node_map = pipeline.get_node_map();
+        let adjacency_list = pipeline.build_adjacency_list(&node_map).unwrap();
+        assert_eq!(adjacency_list.len(), 1);
+        assert_eq!(adjacency_list.get("1").unwrap(), &vec!["2".to_string()]);
+    }
+
+    #[test]
+    fn test_pipeline_build_adjacency_list_invalid_source() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![Node::new(
+                "1".to_string(),
+                NodeData::Stream(StreamParams::new(
+                    "test_org",
+                    "test_stream",
+                    StreamType::Logs,
+                )),
+                100.0,
+                100.0,
+                "input".to_string(),
+            )],
+            edges: vec![Edge {
+                id: "e1-2".to_string(),
+                source: "invalid_source".to_string(), // Invalid source
+                target: "1".to_string(),
+            }],
+        };
+
+        let node_map = pipeline.get_node_map();
+        let result = pipeline.build_adjacency_list(&node_map);
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("source node not found")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_num_of_func() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Function(components::FunctionParams {
+                        name: "test_function".to_string(),
+                        after_flatten: false,
+                        num_args: 0,
+                    }),
+                    200.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "3".to_string(),
+                    NodeData::Function(components::FunctionParams {
+                        name: "test_function2".to_string(),
+                        after_flatten: true,
+                        num_args: 0,
+                    }),
+                    300.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+            ],
+            edges: vec![],
+        };
+
+        assert_eq!(pipeline.num_of_func(), 2);
+    }
+
+    #[test]
+    fn test_pipeline_contains_function() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Function(components::FunctionParams {
+                        name: "test_function".to_string(),
+                        after_flatten: false,
+                        num_args: 0,
+                    }),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    200.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+            ],
+            edges: vec![],
+        };
+
+        assert!(pipeline.contains_function("test_function"));
+        assert!(!pipeline.contains_function("nonexistent_function"));
+    }
+
+    #[test]
+    fn test_pipeline_get_source_stream_params_realtime() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![],
+            edges: vec![],
+        };
+
+        let stream_params = pipeline.get_source_stream_params();
+        assert_eq!(stream_params.org_id, "test_org");
+        assert_eq!(stream_params.stream_name, "test_stream");
+        assert_eq!(stream_params.stream_type, StreamType::Logs);
+    }
+
+    #[test]
+    fn test_pipeline_get_source_stream_params_scheduled() {
+        let derived_stream = DerivedStream {
+            org_id: "test_org".to_string(),
+            stream_type: StreamType::Logs,
+            query_condition: QueryCondition {
+                sql: Some("SELECT * FROM test_stream".to_string()),
+                ..Default::default()
+            },
+            trigger_condition: TriggerCondition::default(),
+            tz_offset: 0,
+            delay: None,
+            start_at: None,
+        };
+
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Scheduled(derived_stream),
+            nodes: vec![],
+            edges: vec![],
+        };
+
+        let stream_params = pipeline.get_source_stream_params();
+        assert_eq!(stream_params.org_id, "test_org");
+        assert_eq!(stream_params.stream_name, "test_stream");
+        assert_eq!(stream_params.stream_type, StreamType::Logs);
+    }
+
+    #[test]
+    fn test_pipeline_contains_remote_destination() {
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![Node::new(
+                "1".to_string(),
+                NodeData::RemoteStream(crate::meta::stream::RemoteStreamParams {
+                    org_id: "test_org".to_string().into(),
+                    destination_name: "test_destination".to_string().into(),
+                }),
+                100.0,
+                100.0,
+                "input".to_string(),
+            )],
+            edges: vec![],
+        };
+
+        assert!(pipeline.contains_remote_destination("test_destination"));
+        assert!(!pipeline.contains_remote_destination("nonexistent_destination"));
+    }
+
+    #[test]
+    fn test_pipeline_get_metadata_by_stream_params() {
+        let mut node = Node::new(
+            "1".to_string(),
+            NodeData::Stream(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            100.0,
+            100.0,
+            "input".to_string(),
+        );
+        node.meta = Some(HashMap::from([
+            ("key1".to_string(), "value1".to_string()),
+            ("key2".to_string(), "value2".to_string()),
+        ]));
+
+        let pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![node],
+            edges: vec![],
+        };
+
+        let stream_params = StreamParams::new("test_org", "test_stream", StreamType::Logs);
+        let metadata = pipeline.get_metadata_by_stream_params(&stream_params);
+        assert!(metadata.is_some());
+        let metadata = metadata.unwrap();
+        assert_eq!(metadata.get("key1"), Some(&"value1".to_string()));
+        assert_eq!(metadata.get("key2"), Some(&"value2".to_string()));
+
+        // Test with non-matching stream params
+        let non_matching_params =
+            StreamParams::new("test_org", "different_stream", StreamType::Logs);
+        let metadata = pipeline.get_metadata_by_stream_params(&non_matching_params);
+        assert!(metadata.is_none());
+    }
+
+    #[test]
+    fn test_pipeline_dependency_item() {
+        let item = PipelineDependencyItem {
+            id: "test_id".to_string(),
+            name: "test_name".to_string(),
+        };
+
+        assert_eq!(item.id, "test_id");
+        assert_eq!(item.name, "test_name");
+    }
+
+    #[test]
+    fn test_pipeline_dependency_response() {
+        let response = PipelineDependencyResponse {
+            list: vec![
+                PipelineDependencyItem {
+                    id: "test_id1".to_string(),
+                    name: "test_name1".to_string(),
+                },
+                PipelineDependencyItem {
+                    id: "test_id2".to_string(),
+                    name: "test_name2".to_string(),
+                },
+            ],
+        };
+
+        assert_eq!(response.list.len(), 2);
+        assert_eq!(response.list[0].id, "test_id1");
+        assert_eq!(response.list[1].name, "test_name2");
+    }
+
+    #[test]
+    fn test_default_status() {
+        assert!(default_status());
+    }
+
+    #[test]
+    fn test_pipeline_validation_cyclical_detection() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream",
+                        StreamType::Logs,
+                    )),
+                    300.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+            ],
+            edges: vec![
+                Edge {
+                    id: "e1-2".to_string(),
+                    source: "1".to_string(),
+                    target: "2".to_string(),
+                },
+                Edge {
+                    id: "e2-1".to_string(),
+                    source: "2".to_string(),
+                    target: "1".to_string(), // Creates a cycle
+                },
+            ],
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("Cyclical pipeline detected")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_validation_leaf_nodes_must_be_stream() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Function(components::FunctionParams {
+                        name: "test_function".to_string(),
+                        after_flatten: false,
+                        num_args: 0,
+                    }),
+                    300.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+            ],
+            edges: vec![Edge {
+                id: "e1-2".to_string(),
+                source: "1".to_string(),
+                target: "2".to_string(),
+            }],
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("All leaf nodes must be StreamNode")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_validation_after_flatten_rule() {
+        let mut pipeline = Pipeline {
+            id: "test_pipeline".to_string(),
+            version: 1,
+            enabled: true,
+            org: "test_org".to_string(),
+            name: "test_pipeline".to_string(),
+            description: "test description".to_string(),
+            source: PipelineSource::Realtime(StreamParams::new(
+                "test_org",
+                "test_stream",
+                StreamType::Logs,
+            )),
+            nodes: vec![
+                Node::new(
+                    "1".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "test_stream",
+                        StreamType::Logs,
+                    )),
+                    100.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "2".to_string(),
+                    NodeData::Function(components::FunctionParams {
+                        name: "test_function1".to_string(),
+                        after_flatten: true, // Checked
+                        num_args: 0,
+                    }),
+                    200.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "3".to_string(),
+                    NodeData::Function(components::FunctionParams {
+                        name: "test_function2".to_string(),
+                        after_flatten: false, // Unchecked after checked
+                        num_args: 0,
+                    }),
+                    300.0,
+                    100.0,
+                    "input".to_string(),
+                ),
+                Node::new(
+                    "4".to_string(),
+                    NodeData::Stream(StreamParams::new(
+                        "test_org",
+                        "output_stream",
+                        StreamType::Logs,
+                    )),
+                    400.0,
+                    100.0,
+                    "output".to_string(),
+                ),
+            ],
+            edges: vec![
+                Edge {
+                    id: "e1-2".to_string(),
+                    source: "1".to_string(),
+                    target: "2".to_string(),
+                },
+                Edge {
+                    id: "e2-3".to_string(),
+                    source: "2".to_string(),
+                    target: "3".to_string(),
+                },
+                Edge {
+                    id: "e3-4".to_string(),
+                    source: "3".to_string(),
+                    target: "4".to_string(),
+                },
+            ],
+        };
+
+        let result = pipeline.validate();
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("After Flatten must be checked")
+        );
+    }
 }

src/config/src/meta/sql.rs

@@ -1111,8 +1111,7 @@ mod tests {
     fn parse_sql_works() {
         let table = "index.1.2022";
         let sql = format!(
-            "select a, b, c from \"{}\" where a=1 and b=1 or c=1 order by c desc limit 5 offset 10",
-            table
+            "select a, b, c from \"{table}\" where a=1 and b=1 or c=1 order by c desc limit 5 offset 10"
         );
         let dialect = sqlparser::dialect::GenericDialect {};
         let statement = &Parser::parse_sql(&dialect, sql.as_ref()).unwrap()[0];
@@ -1129,8 +1128,7 @@ mod tests {
     fn test_sql_new() {
         let table = "index.1.2022";
         let sql = format!(
-            "select a, b, c from \"{}\" where a=1 and b=1 or c=1 order by c desc limit 5 offset 10",
-            table
+            "select a, b, c from \"{table}\" where a=1 and b=1 or c=1 order by c desc limit 5 offset 10"
         );
         let local_sql: Sql = Sql::new(sql.as_str()).unwrap();
         assert_eq!(local_sql.source, table);
@@ -1284,7 +1282,7 @@ mod tests {
     fn test_resolve_stream_names_with_type() {
         let sql = "select * from \"log\".default";
         let names = resolve_stream_names_with_type(sql).unwrap();
-        println!("{:?}", names);
+        println!("{names:?}");
     }
 
     #[test]

src/config/src/meta/stream.rs

@@ -1346,7 +1346,7 @@ mod tests {
         let other = TimeRange::new(50, 100);
         let (left, right) = range.split_by_range(&other).unwrap();
         assert!(left.as_ref().is_some());
-        assert!(!right.is_some());
+        assert!(right.is_none());
         assert_eq!(left.unwrap(), TimeRange::new(0, 50));
 
         // equals

src/config/src/utils/enrichment_local_cache.rs

@@ -91,3 +91,70 @@ pub async fn get_metadata_content() -> Result<HashMap<String, i64>> {
         Ok(HashMap::new())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_get_key() {
+        let key = get_key("org123", "table456");
+        assert_eq!(key, "org123/table456");
+
+        let key2 = get_key("test_org", "my_table");
+        assert_eq!(key2, "test_org/my_table");
+
+        // Test with special characters
+        let key3 = get_key("test-org@123", "table_name_with_spaces");
+        assert_eq!(key3, "test-org@123/table_name_with_spaces");
+
+        // Test with empty strings
+        let key4 = get_key("", "");
+        assert_eq!(key4, "/");
+    }
+
+    #[test]
+    fn test_get_table_path() {
+        let table_dir = "/tmp/test_cache/org123/table456";
+        let created_at = 1640995200; // 2022-01-01 00:00:00 UTC
+
+        let table_path = get_table_path(table_dir, created_at);
+        let expected_path = format!("{table_dir}/{created_at}.json");
+        assert_eq!(table_path.to_string_lossy(), expected_path);
+
+        // Test with different timestamp
+        let table_path2 = get_table_path(table_dir, 1640995300);
+        let expected_path2 = format!("{}/{}.json", table_dir, 1640995300);
+        assert_eq!(table_path2.to_string_lossy(), expected_path2);
+    }
+
+    #[test]
+    fn test_get_table_dir_and_metadata_path() {
+        // Test that the functions return valid paths
+        let table_dir = get_table_dir("org123/table456");
+        assert!(
+            table_dir
+                .to_string_lossy()
+                .contains("enrichment_table_cache")
+        );
+        assert!(table_dir.to_string_lossy().contains("org123/table456"));
+
+        let metadata_path = get_metadata_path();
+        assert!(
+            metadata_path
+                .to_string_lossy()
+                .contains("enrichment_table_cache")
+        );
+        assert!(metadata_path.to_string_lossy().contains("metadata.json"));
+    }
+
+    #[test]
+    fn test_get_table_dir_with_custom_cache_dir() {
+        // Test that the functions return valid paths with custom cache dir
+        let table_dir = get_table_dir("org123/table456");
+        assert!(table_dir.to_string_lossy().contains("org123/table456"));
+
+        let metadata_path = get_metadata_path();
+        assert!(metadata_path.to_string_lossy().contains("metadata.json"));
+    }
+}

src/config/src/utils/file.rs

@@ -193,7 +193,7 @@ mod tests {
         );
 
         std::fs::remove_file(&file_name).unwrap();
-        std::fs::remove_dir(&dir_name).unwrap();
+        std::fs::remove_dir(dir_name).unwrap();
     }
 
     #[test]

src/config/src/utils/flatten.rs

@@ -232,7 +232,7 @@ mod tests {
         assert_eq!(
             flatten(obj).unwrap(),
             json!({
-                format!("s{k}a", k = KEY_SEPARATOR): "[1,2.0,\"b\",null,true]",
+                format!("s{KEY_SEPARATOR}a"): "[1,2.0,\"b\",null,true]",
             })
         );
     }

src/config/src/utils/hash/cityhash.rs

@@ -38,7 +38,7 @@ mod tests {
             "hello", "world", "foo", "bar", "test", "test1", "test2", "test3",
         ] {
             let sum = h.sum64(key);
-            println!("{}: {}", key, sum);
+            println!("{key}: {sum}");
         }
         assert_eq!(h.sum64("hello"), 13009744463427800296);
         assert_eq!(h.sum64("world"), 16436542438370751598);

src/config/src/utils/hash/fnv.rs

@@ -56,7 +56,7 @@ mod tests {
             "hello", "world", "foo", "bar", "test", "test1", "test2", "test3",
         ] {
             let sum = h.sum64(key);
-            println!("{}: {}", key, sum);
+            println!("{key}: {sum}");
         }
         assert_eq!(h.sum64("hello"), 11831194018420276491);
         assert_eq!(h.sum64("world"), 5717881983045765875);

src/config/src/utils/hash/gxhash.rs

@@ -48,7 +48,7 @@ mod tests {
             "hello", "world", "foo", "bar", "test", "test1", "test2", "test3",
         ] {
             let sum = h.sum64(key);
-            println!("{}: {}", key, sum);
+            println!("{key}: {sum}");
         }
         assert_eq!(h.sum64("hello"), 17199510979973968020);
         assert_eq!(h.sum64("world"), 16083628800851799373);
@@ -68,7 +68,7 @@ mod tests {
             "hello", "world", "foo", "bar", "test", "test1", "test2", "test3",
         ] {
             let sum = h.sum64(key);
-            println!("{}: {}", key, sum);
+            println!("{key}: {sum}");
         }
         assert_eq!(h.sum64("hello"), 16350172494705860510);
         assert_eq!(h.sum64("world"), 17970961829702799988);

src/config/src/utils/hash/murmur3.rs

@@ -44,7 +44,7 @@ mod tests {
             "hello", "world", "foo", "bar", "test", "test1", "test2", "test3",
         ] {
             let sum = h.sum64(key);
-            println!("{}: {}", key, sum);
+            println!("{key}: {sum}");
         }
         assert_eq!(h.sum64("hello"), 14688674573012802306);
         assert_eq!(h.sum64("world"), 8198091784597505258);

src/config/src/utils/json.rs

@@ -268,7 +268,7 @@ mod tests {
             Some(Value::Array(
                 [true, false, true]
                     .into_iter()
-                    .map(|v| Value::Bool(v))
+                    .map(Value::Bool)
                     .collect_vec()
             ))
         );
@@ -279,7 +279,7 @@ mod tests {
         assert_eq!(
             get_value_from_path(&val, ".*.a"),
             Some(Value::Array(
-                [true,].into_iter().map(|v| Value::Bool(v)).collect_vec()
+                [true,].into_iter().map(Value::Bool).collect_vec()
             ))
         );
 
@@ -291,7 +291,7 @@ mod tests {
             Some(Value::Array(
                 [true, false, true]
                     .into_iter()
-                    .map(|v| Value::Bool(v))
+                    .map(Value::Bool)
                     .collect_vec()
             ))
         );

src/config/src/utils/parquet.rs

@@ -266,7 +266,7 @@ mod tests {
         // Write to parquet
         let data = write_recordbatch_to_parquet(
             schema.clone(),
-            &[batch.clone()],
+            std::slice::from_ref(&batch),
             &["name".to_string()],
             &metadata,
         )

src/config/src/utils/query_select_utils.rs

@@ -336,7 +336,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(username, 'admin')"));
         assert!(!result.contains("::_o2_custom"));
@@ -351,7 +351,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(username, 'admin')"));
         assert!(result.contains("str_match(role, 'superuser')"));
@@ -368,7 +368,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(department, 'engineering')"));
         assert!(result.contains("str_match(sub.name, 'john')"));
@@ -387,7 +387,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(status, 'active')"));
         assert!(result.contains("str_match(role, 'admin')"));
@@ -405,7 +405,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(u.department, 'sales')"));
         assert!(result.contains("str_match(p.category, 'internal')"));
@@ -419,7 +419,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("::_o2_custom")); // Original should remain
     }
@@ -432,7 +432,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("::_o2_custom")); // Original should remain
     }
@@ -471,7 +471,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(username, 'admin')"));
         assert!(!result.contains("::_o2_custom"));
@@ -486,7 +486,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(LOWER(username), 'admin')"));
         assert!(result.contains("str_match(table.field, 'value')"));
@@ -502,7 +502,7 @@ mod tests {
         "#;
 
         let result = replace_o2_custom_patterns(sql).unwrap();
-        println!("result: {}", result);
+        println!("result: {result}");
 
         assert!(result.contains("str_match(username, 'admin')"));
         assert!(result.contains("str_match(role, 'superuser')"));

src/config/src/utils/sort.rs

@@ -48,7 +48,7 @@ mod tests {
 
     #[test]
     fn test_sort_float_array() {
-        let mut numbers = vec![
+        let mut numbers = [
             1.5,
             f64::NAN,
             0.0,
@@ -56,7 +56,7 @@ mod tests {
             -2.7,
             f64::NEG_INFINITY,
             f64::NAN,
-            3.14,
+            std::f64::consts::PI,
         ];
 
         // Sort the vector using our custom sort_float function
@@ -67,14 +67,14 @@ mod tests {
         // Then negative infinity
         // Then regular numbers in ascending order
         // Then positive infinity
-        let expected = vec![
+        let expected = [
             f64::NAN,
             f64::NAN,
             f64::NEG_INFINITY,
             -2.7,
             0.0,
             1.5,
-            3.14,
+            std::f64::consts::PI,
             f64::INFINITY,
         ];
 
@@ -86,8 +86,7 @@ mod tests {
             }
             assert_eq!(
                 actual, expected,
-                "Mismatch at position {}: expected {:?}, got {:?}",
-                i, expected, actual
+                "Mismatch at position {i}: expected {expected:?}, got {actual:?}"
             );
         }
     }

src/config/src/utils/sql.rs

@@ -698,8 +698,7 @@ mod tests {
             let result = is_timestamp_selected(sql)?;
             assert_eq!(
                 result, expected,
-                "Failed test case '{}': expected {}, got {}",
-                test_name, expected, result
+                "Failed test case '{test_name}': expected {expected}, got {result}"
             );
         }
 
@@ -710,7 +709,7 @@ mod tests {
     fn check_is_simple_aggregate() {
         let query = r#"SELECT histogram(_timestamp) AS zo_sql_time, "kubernetes_docker_id" AS zo_sql_key, SUM(count) AS zo_sql_num FROM "distinct_values_logs_default22" GROUP BY zo_sql_time, zo_sql_key ORDER BY zo_sql_time ASC, zo_sql_num DESC"#;
         let ab = is_aggregate_query(query);
-        print!("{:?}", ab);
+        print!("{ab:?}");
     }
 
     #[test]
@@ -868,13 +867,11 @@ mod tests {
         for (i, (query, description)) in queries.iter().enumerate() {
             let is_simple_aggregate = is_simple_aggregate_query(query).unwrap();
             println!(
-                "Query [{}]: {} - is_simple: {:?}",
-                i, description, is_simple_aggregate
+                "Query [{i}]: {description} - is_simple: {is_simple_aggregate:?}"
             );
-            assert_eq!(
-                is_simple_aggregate, false,
-                "Failed test case [{}]: '{}' - should not be simple but returned true",
-                i, description
+            assert!(
+                !is_simple_aggregate,
+                "Failed test case [{i}]: '{description}' - should not be simple but returned true"
             );
         }
     }
@@ -944,13 +941,11 @@ mod tests {
         for (i, (query, description)) in queries.iter().enumerate() {
             let is_simple_aggregate = is_simple_aggregate_query(query).unwrap();
             println!(
-                "Simple Query [{}]: {} - is_simple: {:?}",
-                i, description, is_simple_aggregate
+                "Simple Query [{i}]: {description} - is_simple: {is_simple_aggregate:?}"
             );
-            assert_eq!(
-                is_simple_aggregate, true,
-                "Failed test case [{}]: '{}' - should be simple but returned false",
-                i, description
+            assert!(
+                is_simple_aggregate,
+                "Failed test case [{i}]: '{description}' - should be simple but returned false"
             );
         }
     }
@@ -1007,8 +1002,8 @@ mod tests {
 
         for (i, query) in queries.iter().enumerate() {
             let is_simple_aggregate = is_simple_aggregate_query(query).unwrap();
-            println!("Query [{}] is_simple: {:?}", i, is_simple_aggregate);
-            assert_eq!(is_simple_aggregate, false);
+            println!("Query [{i}] is_simple: {is_simple_aggregate:?}");
+            assert!(!is_simple_aggregate);
         }
     }
 }

src/config/src/utils/time.rs

@@ -436,7 +436,7 @@ mod tests {
 
         // Test with input 0 (uses current time, so we can't test the exact value)
         let result = get_ymdh_from_micros(0);
-        assert!(result.len() > 0);
+        assert!(!result.is_empty());
     }
 
     #[test]

src/handler/http/request/authz/fga.rs

@@ -415,7 +415,7 @@ pub async fn get_users_with_role(path: web::Path<(String, String)>) -> Result<Ht
     )
 )]
 #[get("/{org_id}/roles/{role_id}/users")]
-pub async fn get_users_with_role(_org_id: web::Path<String>) -> Result<HttpResponse, Error> {
+pub async fn get_users_with_role(_path: web::Path<(String, String)>) -> Result<HttpResponse, Error> {
     Ok(MetaHttpResponse::forbidden("Not Supported"))
 }
 
@@ -584,7 +584,7 @@ pub async fn update_group(
 )]
 #[put("/{org_id}/groups/{group_name}")]
 pub async fn update_group(
-    _org_id: web::Path<String>,
+    _path: web::Path<(String, String)>,
     _user_group: web::Json<UserGroupRequest>,
 ) -> Result<HttpResponse, Error> {
     Ok(MetaHttpResponse::forbidden("Not Supported"))
@@ -803,3 +803,541 @@ pub async fn delete_group(path: web::Path<(String, String)>) -> Result<HttpRespo
 pub async fn delete_group(_path: web::Path<(String, String)>) -> Result<HttpResponse, Error> {
     Ok(MetaHttpResponse::forbidden("Not Supported"))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashSet;
+
+    use actix_web::{App, test};
+
+    use super::*;
+    use crate::common::meta::user::{UserGroup, UserGroupRequest, UserRoleRequest};
+
+    #[tokio::test]
+    async fn test_create_role_enterprise_success() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(create_role)).await;
+            let role_request = UserRoleRequest {
+                role: "custom_role".to_string(),
+                custom: None,
+            };
+
+            let req = test::TestRequest::post()
+                .uri("/test_org/roles")
+                .set_json(&role_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Note: This will likely fail in test environment due to missing OFGA setup
+            // but we're testing the endpoint structure and request handling
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(create_role)).await;
+            let role_request = UserRoleRequest {
+                role: "custom_role".to_string(),
+                custom: None,
+            };
+
+            let req = test::TestRequest::post()
+                .uri("/test_org/roles")
+                .set_json(&role_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_create_role_empty_name() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(create_role)).await;
+            let role_request = UserRoleRequest {
+                role: "   ".to_string(), // Empty after trim
+                custom: None,
+            };
+
+            let req = test::TestRequest::post()
+                .uri("/test_org/roles")
+                .set_json(&role_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 400); // Bad request for empty role name
+        }
+    }
+
+    #[tokio::test]
+    async fn test_create_role_standard_role() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(create_role)).await;
+            let role_request = UserRoleRequest {
+                role: "admin".to_string(), // Standard role
+                custom: None,
+            };
+
+            let req = test::TestRequest::post()
+                .uri("/test_org/roles")
+                .set_json(&role_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 400); // Bad request for standard role
+        }
+    }
+
+    #[tokio::test]
+    async fn test_delete_role_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(delete_role)).await;
+            let req = test::TestRequest::delete()
+                .uri("/test_org/roles/test_role")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(delete_role)).await;
+            let req = test::TestRequest::delete()
+                .uri("/test_org/roles/test_role")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_roles_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_roles)).await;
+            let mut req = test::TestRequest::get().uri("/test_org/roles").to_request();
+
+            // Add user_id header that the function expects
+            req.headers_mut()
+                .insert("user_id", "test_user".parse().unwrap());
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_roles)).await;
+            let req = test::TestRequest::get().uri("/test_org/roles").to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_update_role_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(update_role)).await;
+            let role_request = o2_dex::meta::auth::RoleRequest {
+                add: Some(vec!["permission1".to_string()]),
+                remove: Some(vec!["permission2".to_string()]),
+                add_users: Some(vec!["user1".to_string()]),
+                remove_users: Some(vec!["user2".to_string()]),
+            };
+
+            let req = test::TestRequest::put()
+                .uri("/test_org/roles/test_role")
+                .set_json(&role_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(update_role)).await;
+            let req = test::TestRequest::post()
+                .uri("/test_org/roles/test_role")
+                .set_json(&"test")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_role_permissions_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_role_permissions)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/roles/test_role/permissions/test_resource")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_role_permissions)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/roles/test_role/permissions/test_resource")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_users_with_role_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_users_with_role)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/roles/test_role/users")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_users_with_role)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/roles/test_role/users")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_roles_for_user_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_roles_for_user)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/users/test@example.com/roles")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_roles_for_user)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/users/test@example.com/roles")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_groups_for_user_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_groups_for_user)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/users/test@example.com/groups")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_groups_for_user)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/users/test@example.com/groups")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_create_group_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(create_group)).await;
+            let mut users = HashSet::new();
+            users.insert("user1".to_string());
+            users.insert("user2".to_string());
+
+            let group = UserGroup {
+                name: "test_group".to_string(),
+                users: Some(users),
+                roles: None,
+            };
+
+            let req = test::TestRequest::post()
+                .uri("/test_org/groups")
+                .set_json(&group)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(create_group)).await;
+            let group = UserGroup {
+                name: "test_group".to_string(),
+                users: None,
+                roles: None,
+            };
+
+            let req = test::TestRequest::post()
+                .uri("/test_org/groups")
+                .set_json(&group)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_update_group_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(update_group)).await;
+            let mut add_users = HashSet::new();
+            add_users.insert("user1".to_string());
+
+            let group_request = UserGroupRequest {
+                add_users: Some(add_users),
+                remove_users: None,
+                add_roles: None,
+                remove_roles: None,
+            };
+
+            let req = test::TestRequest::put()
+                .uri("/test_org/groups/test_group")
+                .set_json(&group_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(update_group)).await;
+            let group_request = UserGroupRequest {
+                add_users: None,
+                remove_users: None,
+                add_roles: None,
+                remove_roles: None,
+            };
+
+            let req = test::TestRequest::put()
+                .uri("/test_org/groups/test_group")
+                .set_json(&group_request)
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_groups_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_groups)).await;
+            let mut req = test::TestRequest::get()
+                .uri("/test_org/groups")
+                .to_request();
+
+            // Add user_id header that the function expects
+            req.headers_mut()
+                .insert("user_id", "test_user".parse().unwrap());
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_groups)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/groups")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_group_details_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_group_details)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/groups/test_group")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_group_details)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/groups/test_group")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_delete_group_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(delete_group)).await;
+            let req = test::TestRequest::delete()
+                .uri("/test_org/groups/test_group")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Will likely fail due to missing OFGA setup, but testing structure
+            assert!(resp.status().is_client_error() || resp.status().is_server_error());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(delete_group)).await;
+            let req = test::TestRequest::delete()
+                .uri("/test_org/groups/test_group")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_get_resources_enterprise() {
+        #[cfg(feature = "enterprise")]
+        {
+            let app = test::init_service(App::new().service(get_resources)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/resources")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            // Should succeed as it doesn't require OFGA setup
+            assert!(resp.status().is_success());
+        }
+
+        #[cfg(not(feature = "enterprise"))]
+        {
+            let app = test::init_service(App::new().service(get_resources)).await;
+            let req = test::TestRequest::get()
+                .uri("/test_org/resources")
+                .to_request();
+
+            let resp = test::call_service(&app, req).await;
+            assert_eq!(resp.status(), 403); // Forbidden in non-enterprise
+        }
+    }
+
+    #[tokio::test]
+    async fn test_user_role_request_validation() {
+        let valid_request = UserRoleRequest {
+            role: "custom_role".to_string(),
+            custom: Some(vec!["permission1".to_string(), "permission2".to_string()]),
+        };
+        assert_eq!(valid_request.role, "custom_role");
+        assert_eq!(
+            valid_request.custom,
+            Some(vec!["permission1".to_string(), "permission2".to_string()])
+        );
+
+        let simple_request = UserRoleRequest {
+            role: "simple_role".to_string(),
+            custom: None,
+        };
+        assert_eq!(simple_request.role, "simple_role");
+        assert!(simple_request.custom.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_user_group_validation() {
+        let mut users = HashSet::new();
+        users.insert("user1".to_string());
+        users.insert("user2".to_string());
+
+        let mut roles = HashSet::new();
+        roles.insert("role1".to_string());
+
+        let group = UserGroup {
+            name: "test_group".to_string(),
+            users: Some(users.clone()),
+            roles: Some(roles.clone()),
+        };
+
+        assert_eq!(group.name, "test_group");
+        assert_eq!(group.users, Some(users));
+        assert_eq!(group.roles, Some(roles));
+    }
+
+    #[tokio::test]
+    async fn test_user_group_request_validation() {
+        let mut add_users = HashSet::new();
+        add_users.insert("user1".to_string());
+
+        let mut remove_users = HashSet::new();
+        remove_users.insert("user2".to_string());
+
+        let request = UserGroupRequest {
+            add_users: Some(add_users.clone()),
+            remove_users: Some(remove_users.clone()),
+            add_roles: None,
+            remove_roles: None,
+        };
+
+        assert_eq!(request.add_users, Some(add_users));
+        assert_eq!(request.remove_users, Some(remove_users));
+        assert!(request.add_roles.is_none());
+        assert!(request.remove_roles.is_none());
+    }
+}

src/infra/src/table/migration/m20250320_000001_remove_alert_name_unique_constraint.rs

@@ -190,3 +190,104 @@ async fn update_scheduled_triggers(manager: &SchemaManager<'_>) -> Result<(), Db
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use sea_orm_migration::prelude::*;
+
+    use super::*;
+
+    #[test]
+    fn test_scheduled_jobs_enum() {
+        assert_eq!(ScheduledJobs::Table.to_string(), "scheduled_jobs");
+        assert_eq!(ScheduledJobs::Org.to_string(), "org");
+        assert_eq!(ScheduledJobs::Module.to_string(), "module");
+        assert_eq!(ScheduledJobs::ModuleKey.to_string(), "module_key");
+    }
+
+    #[test]
+    fn test_alerts_enum() {
+        assert_eq!(Alerts::Table.to_string(), "alerts");
+        assert_eq!(Alerts::Id.to_string(), "id");
+        assert_eq!(Alerts::Org.to_string(), "org");
+        assert_eq!(Alerts::StreamType.to_string(), "stream_type");
+        assert_eq!(Alerts::StreamName.to_string(), "stream_name");
+        assert_eq!(Alerts::Name.to_string(), "name");
+    }
+
+    #[test]
+    fn test_migration_name() {
+        let migration = Migration;
+        let name = migration.name();
+        assert!(!name.is_empty(), "Migration should have a name");
+        assert!(
+            name.contains("remove_alert_name_unique_constraint"),
+            "Migration name should contain the constraint name"
+        );
+    }
+
+    #[test]
+    fn test_module_key_parsing() {
+        // Test valid module key format
+        let valid_key = "logs/test_stream/test_alert";
+        let parts: Vec<&str> = valid_key.split('/').collect();
+        assert_eq!(parts.len(), 3, "Valid module key should have 3 parts");
+        assert_eq!(parts[0], "logs");
+        assert_eq!(parts[1], "test_stream");
+        assert_eq!(parts[2], "test_alert");
+
+        // Test invalid module key format
+        let invalid_key = "invalid_format";
+        let invalid_parts: Vec<&str> = invalid_key.split('/').collect();
+        assert_ne!(
+            invalid_parts.len(),
+            3,
+            "Invalid module key should not have 3 parts"
+        );
+    }
+
+    #[test]
+    fn test_alert_name_unique_constraint_constant() {
+        // Test that the constant is properly defined
+        assert_eq!(
+            ALERTS_ORG_STREAM_TYPE_STREAM_NAME_NAME_IDX,
+            "alerts_org_stream_type_stream_name_name_idx_2"
+        );
+    }
+
+    #[test]
+    fn test_migration_structure() {
+        let migration = Migration;
+
+        // Test that the migration implements the required trait
+        let name = migration.name();
+        assert!(!name.is_empty());
+
+        // Test that the migration has the expected structure
+        assert!(name.contains("remove_alert_name_unique_constraint"));
+    }
+
+    #[test]
+    fn test_enum_derivation() {
+        // Test that the enums are properly derived
+        let scheduled_jobs_table = ScheduledJobs::Table;
+        let alerts_table = Alerts::Table;
+
+        assert_eq!(scheduled_jobs_table.to_string(), "scheduled_jobs");
+        assert_eq!(alerts_table.to_string(), "alerts");
+    }
+
+    #[test]
+    fn test_migration_trait_implementation() {
+        // Test that the migration implements MigrationTrait
+        let migration = Migration;
+
+        // This test ensures the migration has the required trait implementation
+        // The actual up/down methods are tested in integration tests
+        assert!(
+            migration
+                .name()
+                .contains("remove_alert_name_unique_constraint")
+        );
+    }
+}

src/service/dashboards/reports.rs

@@ -240,7 +240,7 @@ pub async fn list(
     permitted: Option<Vec<String>>,
 ) -> Result<Vec<table::reports::ListReportsQueryResult>, ReportError> {
     let conn = ORM_CLIENT.get_or_init(connect_to_orm).await;
-    let params = filters.into_parmas(org_id);
+    let params = filters.into_params(org_id);
     let reports = db::dashboards::reports::list(conn, &params)
         .await
         .map_err(ReportError::DbError)?;

src/service/search/datafusion/plan/tantivy_optimize_exec.rs

@@ -546,7 +546,9 @@ fn parse_bool_array(field_values: &[String]) -> Result<Arc<dyn Array>, DataFusio
 
 #[cfg(test)]
 mod tests {
-    use arrow::array::Float64Array;
+    use arrow::array::{BooleanArray, Float64Array, Int64Array, StringArray, UInt64Array};
+    use arrow_schema::{DataType, Field, Schema, TimeUnit};
+    use config::meta::stream::{FileMeta, StreamType};
 
     use super::*;
 
@@ -563,4 +565,639 @@ mod tests {
         assert_eq!(array_values.value(3), f64::INFINITY);
         assert_eq!(array_values.value(4), f64::NEG_INFINITY);
     }
+
+    #[test]
+    fn test_parse_i64_array() {
+        let field_values = vec![
+            "123".to_string(),
+            "-456".to_string(),
+            "0".to_string(),
+            "".to_string(),
+        ];
+        let array = parse_i64_array(&field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<Int64Array>().unwrap();
+        assert_eq!(array_values.len(), 4);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), -456);
+        assert_eq!(array_values.value(2), 0);
+        assert_eq!(array_values.value(3), 0); // empty string defaults to 0
+    }
+
+    #[test]
+    fn test_parse_u64_array() {
+        let field_values = vec![
+            "123".to_string(),
+            "456".to_string(),
+            "0".to_string(),
+            "".to_string(),
+        ];
+        let array = parse_u64_array(&field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<UInt64Array>().unwrap();
+        assert_eq!(array_values.len(), 4);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+        assert_eq!(array_values.value(2), 0);
+        assert_eq!(array_values.value(3), 0); // empty string defaults to 0
+    }
+
+    #[test]
+    fn test_parse_bool_array() {
+        let field_values = vec!["true".to_string(), "false".to_string(), "".to_string()];
+        let array = parse_bool_array(&field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<BooleanArray>().unwrap();
+        assert_eq!(array_values.len(), 3);
+        assert!(array_values.value(0));
+        assert!(!array_values.value(1));
+        assert!(!array_values.value(2)); // empty string defaults to false
+    }
+
+    #[test]
+    fn test_parse_i64_array_invalid_input() {
+        let field_values = vec!["123".to_string(), "invalid".to_string()];
+        let result = parse_i64_array(&field_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_parse_u64_array_invalid_input() {
+        let field_values = vec!["123".to_string(), "invalid".to_string()];
+        let result = parse_u64_array(&field_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_parse_f64_array_invalid_input() {
+        let field_values = vec!["123.45".to_string(), "invalid".to_string()];
+        let result = parse_f64_array(&field_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_parse_bool_array_invalid_input() {
+        let field_values = vec!["true".to_string(), "invalid".to_string()];
+        let result = parse_bool_array(&field_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_create_field_array_string() {
+        let field = Field::new("test", DataType::Utf8, false);
+        let field_values = vec!["hello".to_string(), "world".to_string()];
+        let array = create_field_array(&field, field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<StringArray>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), "hello");
+        assert_eq!(array_values.value(1), "world");
+    }
+
+    #[test]
+    fn test_create_field_array_int64() {
+        let field = Field::new("test", DataType::Int64, false);
+        let field_values = vec!["123".to_string(), "456".to_string()];
+        let array = create_field_array(&field, field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<Int64Array>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+    }
+
+    #[test]
+    fn test_create_field_array_uint64() {
+        let field = Field::new("test", DataType::UInt64, false);
+        let field_values = vec!["123".to_string(), "456".to_string()];
+        let array = create_field_array(&field, field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<UInt64Array>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+    }
+
+    #[test]
+    fn test_create_field_array_float64() {
+        let field = Field::new("test", DataType::Float64, false);
+        let field_values = vec!["123.45".to_string(), "456.78".to_string()];
+        let array = create_field_array(&field, field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<Float64Array>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123.45);
+        assert_eq!(array_values.value(1), 456.78);
+    }
+
+    #[test]
+    fn test_create_field_array_boolean() {
+        let field = Field::new("test", DataType::Boolean, false);
+        let field_values = vec!["true".to_string(), "false".to_string()];
+        let array = create_field_array(&field, field_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<BooleanArray>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert!(array_values.value(0));
+        assert!(!array_values.value(1));
+    }
+
+    #[test]
+    fn test_create_field_array_unsupported_type() {
+        let field = Field::new("test", DataType::Int32, false);
+        let field_values = vec!["123".to_string()];
+        let result = create_field_array(&field, field_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_create_count_array_int64() {
+        let field = Field::new("count", DataType::Int64, false);
+        let count_values = vec![123, 456];
+        let array = create_count_array(&field, count_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<Int64Array>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+    }
+
+    #[test]
+    fn test_create_count_array_uint64() {
+        let field = Field::new("count", DataType::UInt64, false);
+        let count_values = vec![123, 456];
+        let array = create_count_array(&field, count_values).unwrap();
+        let array_values = array.as_any().downcast_ref::<UInt64Array>().unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+    }
+
+    #[test]
+    fn test_create_count_array_int32() {
+        let field = Field::new("count", DataType::Int32, false);
+        let count_values = vec![123, 456];
+        let array = create_count_array(&field, count_values).unwrap();
+        let array_values = array
+            .as_any()
+            .downcast_ref::<arrow::array::Int32Array>()
+            .unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+    }
+
+    #[test]
+    fn test_create_count_array_uint32() {
+        let field = Field::new("count", DataType::UInt32, false);
+        let count_values = vec![123, 456];
+        let array = create_count_array(&field, count_values).unwrap();
+        let array_values = array
+            .as_any()
+            .downcast_ref::<arrow::array::UInt32Array>()
+            .unwrap();
+        assert_eq!(array_values.len(), 2);
+        assert_eq!(array_values.value(0), 123);
+        assert_eq!(array_values.value(1), 456);
+    }
+
+    #[test]
+    fn test_create_count_array_unsupported_type() {
+        let field = Field::new("count", DataType::Float64, false);
+        let count_values = vec![123, 456];
+        let result = create_count_array(&field, count_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_create_histogram_arrow_array_microsecond() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "timestamp",
+                DataType::Timestamp(TimeUnit::Microsecond, None),
+                false,
+            ),
+            Field::new("count", DataType::Int64, false),
+        ]));
+        let histogram_counts = vec![10, 0, 20, 0, 30];
+        let result = create_histogram_arrow_array(&schema, histogram_counts, 1000, 100, 5).unwrap();
+
+        assert_eq!(result.len(), 2);
+        let timestamp_array = result[0]
+            .as_any()
+            .downcast_ref::<TimestampMicrosecondArray>()
+            .unwrap();
+        let count_array = result[1].as_any().downcast_ref::<Int64Array>().unwrap();
+
+        assert_eq!(timestamp_array.len(), 3); // Only non-zero counts
+        assert_eq!(count_array.len(), 3);
+        assert_eq!(timestamp_array.value(0), 1000);
+        assert_eq!(timestamp_array.value(1), 1200);
+        assert_eq!(timestamp_array.value(2), 1400);
+        assert_eq!(count_array.value(0), 10);
+        assert_eq!(count_array.value(1), 20);
+        assert_eq!(count_array.value(2), 30);
+    }
+
+    #[test]
+    fn test_create_histogram_arrow_array_nanosecond() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "timestamp",
+                DataType::Timestamp(TimeUnit::Nanosecond, None),
+                false,
+            ),
+            Field::new("count", DataType::UInt64, false),
+        ]));
+        let histogram_counts = vec![10, 20];
+        let result = create_histogram_arrow_array(&schema, histogram_counts, 1000, 100, 2).unwrap();
+
+        assert_eq!(result.len(), 2);
+        let timestamp_array = result[0]
+            .as_any()
+            .downcast_ref::<TimestampNanosecondArray>()
+            .unwrap();
+        let count_array = result[1].as_any().downcast_ref::<UInt64Array>().unwrap();
+
+        assert_eq!(timestamp_array.len(), 2);
+        assert_eq!(count_array.len(), 2);
+        assert_eq!(timestamp_array.value(0), 1000000); // 1000 * 1000
+        assert_eq!(timestamp_array.value(1), 1100000); // 1100 * 1000
+        assert_eq!(count_array.value(0), 10);
+        assert_eq!(count_array.value(1), 20);
+    }
+
+    #[test]
+    fn test_create_histogram_arrow_array_invalid_schema() {
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "timestamp",
+            DataType::Int64,
+            false,
+        )]));
+        let histogram_counts = vec![10, 20];
+        let result = create_histogram_arrow_array(&schema, histogram_counts, 1000, 100, 2);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_create_histogram_arrow_array_zero_bucket_width() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "timestamp",
+                DataType::Timestamp(TimeUnit::Microsecond, None),
+                false,
+            ),
+            Field::new("count", DataType::Int64, false),
+        ]));
+        let histogram_counts = vec![10, 20];
+        let result = create_histogram_arrow_array(&schema, histogram_counts, 1000, 0, 2);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_create_histogram_arrow_array_normalize_counts() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "timestamp",
+                DataType::Timestamp(TimeUnit::Microsecond, None),
+                false,
+            ),
+            Field::new("count", DataType::Int64, false),
+        ]));
+        let histogram_counts = vec![10, 20]; // Only 2 counts
+        let result = create_histogram_arrow_array(&schema, histogram_counts, 1000, 100, 5).unwrap();
+
+        assert_eq!(result.len(), 2);
+        let timestamp_array = result[0]
+            .as_any()
+            .downcast_ref::<TimestampMicrosecondArray>()
+            .unwrap();
+        let count_array = result[1].as_any().downcast_ref::<Int64Array>().unwrap();
+
+        assert_eq!(timestamp_array.len(), 2); // Only non-zero counts
+        assert_eq!(count_array.len(), 2);
+        assert_eq!(timestamp_array.value(0), 1000);
+        assert_eq!(timestamp_array.value(1), 1100);
+        assert_eq!(count_array.value(0), 10);
+        assert_eq!(count_array.value(1), 20);
+    }
+
+    #[test]
+    fn test_create_top_n_arrow_array() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("field", DataType::Utf8, false),
+            Field::new("count", DataType::Int64, false),
+        ]));
+        let top_n = vec![("a".to_string(), 10), ("b".to_string(), 20)];
+        let result = create_top_n_arrow_array(&schema, top_n, "field", 2).unwrap();
+
+        assert_eq!(result.len(), 1); // One batch
+        let batch = &result[0];
+        assert_eq!(batch.len(), 2); // Two arrays
+
+        let field_array = batch[0].as_any().downcast_ref::<StringArray>().unwrap();
+        let count_array = batch[1].as_any().downcast_ref::<Int64Array>().unwrap();
+
+        assert_eq!(field_array.len(), 2);
+        assert_eq!(count_array.len(), 2);
+        assert_eq!(field_array.value(0), "a");
+        assert_eq!(field_array.value(1), "b");
+        assert_eq!(count_array.value(0), 10);
+        assert_eq!(count_array.value(1), 20);
+    }
+
+    #[test]
+    fn test_create_top_n_arrow_array_empty() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("field", DataType::Utf8, false),
+            Field::new("count", DataType::Int64, false),
+        ]));
+        let result = create_top_n_arrow_array(&schema, vec![], "field", 2).unwrap();
+        assert!(result.is_empty());
+    }
+
+    #[test]
+    fn test_create_top_n_arrow_array_invalid_schema() {
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let top_n = vec![("a".to_string(), 10)];
+        let result = create_top_n_arrow_array(&schema, top_n, "field", 1);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_create_distinct_arrow_array() {
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let distinct_values = vec!["a".to_string(), "b".to_string(), "c".to_string()]
+            .into_iter()
+            .collect();
+        let result = create_distinct_arrow_array(&schema, distinct_values).unwrap();
+
+        assert_eq!(result.len(), 1);
+        let field_array = result[0].as_any().downcast_ref::<StringArray>().unwrap();
+        assert_eq!(field_array.len(), 3);
+        // Note: HashSet order is not guaranteed, so we just check the length
+    }
+
+    #[test]
+    fn test_create_distinct_arrow_array_invalid_schema() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("field", DataType::Utf8, false),
+            Field::new("count", DataType::Int64, false),
+        ]));
+        let distinct_values = vec!["a".to_string()].into_iter().collect();
+        let result = create_distinct_arrow_array(&schema, distinct_values);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_tantivy_optimize_exec_new() {
+        let query = Arc::new(QueryParams {
+            trace_id: "test".to_string(),
+            org_id: "org".to_string(),
+            stream_type: StreamType::Logs,
+            stream_name: "test".to_string(),
+            time_range: None,
+            work_group: None,
+            use_inverted_index: false,
+        });
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let file_list = vec![FileKey {
+            id: 1,
+            account: "test_account".to_string(),
+            key: "test.parquet".to_string(),
+            meta: FileMeta::default(),
+            deleted: false,
+            segment_ids: None,
+        }];
+        let index_condition = None;
+        let index_optimize_mode = IndexOptimizeMode::SimpleCount;
+
+        let exec = TantivyOptimizeExec::new(
+            query.clone(),
+            schema.clone(),
+            file_list.clone(),
+            index_condition.clone(),
+            index_optimize_mode.clone(),
+        );
+
+        assert_eq!(exec.query.trace_id, "test");
+        assert_eq!(exec.file_list.len(), 1);
+        assert_eq!(exec.index_optimize_mode, index_optimize_mode);
+    }
+
+    #[test]
+    fn test_tantivy_optimize_exec_display() {
+        let query = Arc::new(QueryParams {
+            trace_id: "test".to_string(),
+            org_id: "org".to_string(),
+            stream_type: StreamType::Logs,
+            stream_name: "test".to_string(),
+            time_range: None,
+            work_group: None,
+            use_inverted_index: false,
+        });
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let file_list = vec![
+            FileKey {
+                id: 1,
+                account: "test_account".to_string(),
+                key: "file1.parquet".to_string(),
+                meta: FileMeta::default(),
+                deleted: false,
+                segment_ids: None,
+            },
+            FileKey {
+                id: 2,
+                account: "test_account".to_string(),
+                key: "file2.parquet".to_string(),
+                meta: FileMeta::default(),
+                deleted: false,
+                segment_ids: None,
+            },
+            FileKey {
+                id: 3,
+                account: "test_account".to_string(),
+                key: "file3.parquet".to_string(),
+                meta: FileMeta::default(),
+                deleted: false,
+                segment_ids: None,
+            },
+            FileKey {
+                id: 4,
+                account: "test_account".to_string(),
+                key: "file4.parquet".to_string(),
+                meta: FileMeta::default(),
+                deleted: false,
+                segment_ids: None,
+            },
+            FileKey {
+                id: 5,
+                account: "test_account".to_string(),
+                key: "file5.parquet".to_string(),
+                meta: FileMeta::default(),
+                deleted: false,
+                segment_ids: None,
+            },
+            FileKey {
+                id: 6,
+                account: "test_account".to_string(),
+                key: "file6.parquet".to_string(),
+                meta: FileMeta::default(),
+                deleted: false,
+                segment_ids: None,
+            },
+        ];
+        let index_condition = None;
+        let index_optimize_mode = IndexOptimizeMode::SimpleCount;
+
+        let exec = Arc::new(TantivyOptimizeExec::new(
+            query,
+            schema,
+            file_list,
+            index_condition,
+            index_optimize_mode,
+        ));
+
+        let display = format!(
+            "{}",
+            datafusion::physical_plan::displayable(exec.as_ref()).indent(false)
+        );
+        assert!(display.contains("TantivyOptimizeExec: files: 6"));
+        assert!(display.contains("optimize_mode: SimpleCount"));
+        assert!(display.contains(
+            "file_list: [file1.parquet, file2.parquet, file3.parquet, file4.parquet, file5.parquet"
+        ));
+    }
+
+    #[test]
+    fn test_tantivy_optimize_exec_execution_plan() {
+        let query = Arc::new(QueryParams {
+            trace_id: "test".to_string(),
+            org_id: "org".to_string(),
+            stream_type: StreamType::Logs,
+            stream_name: "test".to_string(),
+            time_range: None,
+            work_group: None,
+            use_inverted_index: false,
+        });
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let file_list = vec![FileKey {
+            id: 1,
+            account: "test_account".to_string(),
+            key: "test.parquet".to_string(),
+            meta: FileMeta::default(),
+            deleted: false,
+            segment_ids: None,
+        }];
+        let index_condition = None;
+        let index_optimize_mode = IndexOptimizeMode::SimpleCount;
+
+        let exec = TantivyOptimizeExec::new(
+            query,
+            schema,
+            file_list,
+            index_condition,
+            index_optimize_mode,
+        );
+
+        assert_eq!(exec.name(), "TantivyOptimizeExec");
+        assert!(exec.children().is_empty());
+        assert_eq!(exec.properties().output_partitioning().partition_count(), 1);
+    }
+
+    #[test]
+    fn test_tantivy_optimize_exec_with_new_children() {
+        let query = Arc::new(QueryParams {
+            trace_id: "test".to_string(),
+            org_id: "org".to_string(),
+            stream_type: StreamType::Logs,
+            stream_name: "test".to_string(),
+            time_range: None,
+            work_group: None,
+            use_inverted_index: false,
+        });
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let file_list = vec![FileKey {
+            id: 1,
+            account: "test_account".to_string(),
+            key: "test.parquet".to_string(),
+            meta: FileMeta::default(),
+            deleted: false,
+            segment_ids: None,
+        }];
+        let index_condition = None;
+        let index_optimize_mode = IndexOptimizeMode::SimpleCount;
+
+        let exec = Arc::new(TantivyOptimizeExec::new(
+            query,
+            schema,
+            file_list,
+            index_condition,
+            index_optimize_mode,
+        ));
+
+        let result = exec.with_new_children(vec![]);
+        assert!(result.is_ok());
+        let new_exec = result.unwrap();
+        assert_eq!(new_exec.name(), "TantivyOptimizeExec");
+    }
+
+    #[test]
+    fn test_tantivy_optimize_exec_statistics() {
+        let query = Arc::new(QueryParams {
+            trace_id: "test".to_string(),
+            org_id: "org".to_string(),
+            stream_type: StreamType::Logs,
+            stream_name: "test".to_string(),
+            time_range: None,
+            work_group: None,
+            use_inverted_index: false,
+        });
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "field",
+            DataType::Utf8,
+            false,
+        )]));
+        let file_list = vec![FileKey {
+            id: 1,
+            account: "test_account".to_string(),
+            key: "test.parquet".to_string(),
+            meta: FileMeta::default(),
+            deleted: false,
+            segment_ids: None,
+        }];
+        let index_condition = None;
+        let index_optimize_mode = IndexOptimizeMode::SimpleCount;
+
+        let exec = TantivyOptimizeExec::new(
+            query,
+            schema,
+            file_list,
+            index_condition,
+            index_optimize_mode,
+        );
+
+        let stats = exec.partition_statistics(None);
+        assert!(stats.is_ok());
+        let stats = stats.unwrap();
+        assert!(matches!(
+            stats.num_rows,
+            datafusion::common::stats::Precision::Absent
+        ));
+    }
 }

src/wal/tests/wal.rs

@@ -29,14 +29,14 @@ fn test_wal_new() {
     let path = build_file_path(dir, "org", "stream", "1".to_string());
     let (mut writer, _) = Writer::new(path.clone(), 1024_1024, 8 * 1024, None).unwrap();
     for i in 0..entry_num {
-        let data = format!("hello world {}", i);
+        let data = format!("hello world {i}");
         writer.write(data.as_bytes()).unwrap();
     }
     writer.close().unwrap();
 
     let mut reader = Reader::from_path(path).unwrap();
     for i in 0..entry_num {
-        let data = format!("hello world {}", i);
+        let data = format!("hello world {i}");
         let entry = reader.read_entry().unwrap().unwrap();
         assert_eq!(entry, data.as_bytes());
     }
@@ -53,7 +53,7 @@ fn test_wal_build() {
     let path = build_file_path(dir, "org", "stream", "1".to_string());
     let (mut writer, _) = Writer::new(path.clone(), 1024_1024, 8 * 1024, Some(header)).unwrap();
     for i in 0..entry_num {
-        let data = format!("hello world {}", i);
+        let data = format!("hello world {i}");
         writer.write(data.as_bytes()).unwrap();
     }
     writer.close().unwrap();
@@ -64,7 +64,7 @@ fn test_wal_build() {
     assert_eq!(header.get("key2"), Some(&"value2".to_string()));
 
     for i in 0..entry_num {
-        let data = format!("hello world {}", i);
+        let data = format!("hello world {i}");
         let entry = reader.read_entry().unwrap().unwrap();
         assert_eq!(entry, data.as_bytes());
     }
@@ -79,7 +79,7 @@ fn test_position() {
     let path = build_file_path(dir, "org", "stream", "1".to_string());
     let (mut writer, _) = Writer::new(path.clone(), 1024_1024, 8 * 1024, None).unwrap();
     for i in 0..entry_num {
-        let data = format!("hello world {}", i);
+        let data = format!("hello world {i}");
         writer.write(data.as_bytes()).unwrap();
     }
     writer.close().unwrap();
@@ -89,7 +89,7 @@ fn test_position() {
     for _ in 0..entry_num {
         let (_, len) = reader.read_entry_with_length().unwrap();
         pos += wal::ENTRY_HEADER_LEN + len;
-        assert_eq!(pos as u64, reader.current_position().unwrap());
+        assert_eq!(pos, reader.current_position().unwrap());
     }
     assert!(reader.read_entry().unwrap().is_none());
 }
@@ -102,7 +102,7 @@ fn test_metadata() {
     let path = build_file_path(dir, "org", "stream", "1".to_string());
     let (mut writer, _) = Writer::new(path, 1024_1024, 8 * 1024, None).unwrap();
     for i in 0..entry_num {
-        let data = format!("hello world {}", i);
+        let data = format!("hello world {i}");
         writer.write(data.as_bytes()).unwrap();
         sleep(Duration::from_millis(100));
         writer.sync().unwrap();
@@ -131,12 +131,11 @@ fn test_realtime_write_and_read() {
 
     std::thread::spawn(move || {
         for i in 0..entry_num {
-            let data = format!("hello world {}", i);
+            let data = format!("hello world {i}");
             writer.write(data.as_bytes()).unwrap();
             writer.sync().unwrap();
             println!(
-                "write data: {}, pos:{:?}, file_path: {}",
-                data,
+                "write data: {data}, pos:{:?}, file_path: {}",
                 writer.current_position(),
                 writer.path().display()
             );
@@ -186,7 +185,7 @@ fn test_reader_from_path_position_and_metadata() {
     let (mut writer, _) = Writer::new(path.clone(), 1024_1024, 8 * 1024, None).unwrap();
 
     for i in 0..entry_num {
-        let data = format!("hello {}", i);
+        let data = format!("hello {i}");
         writer.write(data.as_bytes()).unwrap();
     }
     writer.close().unwrap();
@@ -209,7 +208,7 @@ fn test_reader_from_path_position_and_metadata() {
 
     for i in 0..entry_num {
         let entry = reader.read_entry().unwrap().unwrap();
-        assert_eq!(entry, format!("hello {}", i).as_bytes());
+        assert_eq!(entry, format!("hello {i}").as_bytes());
     }
     assert!(reader.read_entry().unwrap().is_none());