arrow-graph 0.2.0

#[cfg(test)]
mod tests {
    use crate::algorithms::pathfinding::{ShortestPath, AllPaths};
    use crate::algorithms::centrality::PageRank;
    use crate::algorithms::components::{WeaklyConnectedComponents, StronglyConnectedComponents};
    use crate::algorithms::community::LeidenCommunityDetection;
    use crate::algorithms::aggregation::{TriangleCount, ClusteringCoefficient};
    use crate::{GraphAlgorithm, AlgorithmParams};
    use crate::ArrowGraph;
    use arrow::record_batch::RecordBatch;
    use arrow::array::{StringArray, Float64Array, UInt32Array, UInt64Array};
    use arrow::datatypes::{DataType, Field, Schema};
    use std::sync::Arc;

    fn create_test_graph() -> ArrowGraph {
        let schema = Arc::new(Schema::new(vec![
            Field::new("source", DataType::Utf8, false),
            Field::new("target", DataType::Utf8, false),
            Field::new("weight", DataType::Float64, true),
        ]));

        let source_array = StringArray::from(vec!["A", "A", "B", "C", "D"]);
        let target_array = StringArray::from(vec!["B", "C", "C", "D", "E"]);
        let weight_array = Float64Array::from(vec![
            Some(1.0), 
            Some(4.0), 
            Some(2.0), 
            Some(1.0),
            Some(1.0)
        ]);

        let edges = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(source_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(target_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(weight_array) as Arc<dyn arrow::array::Array>
            ],
        ).unwrap();

        ArrowGraph::from_edges(edges).unwrap()
    }

    fn create_disconnected_graph() -> ArrowGraph {
        let schema = Arc::new(Schema::new(vec![
            Field::new("source", DataType::Utf8, false),
            Field::new("target", DataType::Utf8, false),
            Field::new("weight", DataType::Float64, true),
        ]));

        // Create two disconnected components: A-B and C-D
        let source_array = StringArray::from(vec!["A", "C"]);
        let target_array = StringArray::from(vec!["B", "D"]);
        let weight_array = Float64Array::from(vec![Some(1.0), Some(1.0)]);

        let edges = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(source_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(target_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(weight_array) as Arc<dyn arrow::array::Array>
            ],
        ).unwrap();

        ArrowGraph::from_edges(edges).unwrap()
    }

    #[test]
    fn test_shortest_path_single_target() {
        let graph = create_test_graph();
        let algorithm = ShortestPath;
        
        let params = AlgorithmParams::new()
            .with_param("source", "A")
            .with_param("target", "D");
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert_eq!(result.num_rows(), 1);
        assert_eq!(result.num_columns(), 4);
        
        // Check that we got a result with source A, target D
        let source_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let target_col = result.column(1).as_any().downcast_ref::<StringArray>().unwrap();
        let distance_col = result.column(2).as_any().downcast_ref::<Float64Array>().unwrap();
        
        assert_eq!(source_col.value(0), "A");
        assert_eq!(target_col.value(0), "D");
        assert_eq!(distance_col.value(0), 4.0); // A->B->C->D = 1+2+1 = 4, A->C->D = 4+1 = 5, so shortest is A->B->C->D = 4
    }

    #[test]
    fn test_shortest_path_all_targets() {
        let graph = create_test_graph();
        let algorithm = ShortestPath;
        
        let params = AlgorithmParams::new()
            .with_param("source", "A");
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Should have paths to all reachable nodes except source
        assert!(result.num_rows() > 0);
        assert_eq!(result.num_columns(), 3);
        
        let source_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        
        // All sources should be "A"
        for i in 0..result.num_rows() {
            assert_eq!(source_col.value(i), "A");
        }
    }

    #[test]
    fn test_all_paths() {
        let graph = create_test_graph();
        let algorithm = AllPaths;
        
        let params = AlgorithmParams::new()
            .with_param("source", "A")
            .with_param("target", "D")
            .with_param("max_hops", 5);
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Should find multiple paths from A to D
        assert!(result.num_rows() >= 1);
        assert_eq!(result.num_columns(), 4);
        
        let source_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let target_col = result.column(1).as_any().downcast_ref::<StringArray>().unwrap();
        
        for i in 0..result.num_rows() {
            assert_eq!(source_col.value(i), "A");
            assert_eq!(target_col.value(i), "D");
        }
    }

    #[test]
    fn test_shortest_path_no_path() {
        let graph = create_test_graph();
        let algorithm = ShortestPath;
        
        let params = AlgorithmParams::new()
            .with_param("source", "E")
            .with_param("target", "A");
        
        let result = algorithm.execute(&graph, &params);
        
        // Should return an error since there's no path from E to A
        assert!(result.is_err());
    }

    #[test]
    fn test_pagerank() {
        let graph = create_test_graph();
        let algorithm = PageRank;
        
        let params = AlgorithmParams::new()
            .with_param("damping_factor", 0.85)
            .with_param("max_iterations", 100)
            .with_param("tolerance", 1e-6);
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Should have PageRank scores for all nodes
        assert_eq!(result.num_rows(), graph.node_count());
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let score_col = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        
        // Verify all nodes are present
        let mut found_nodes = std::collections::HashSet::new();
        let mut total_score = 0.0;
        
        for i in 0..result.num_rows() {
            let node_id = node_col.value(i);
            let score = score_col.value(i);
            
            found_nodes.insert(node_id);
            total_score += score;
            
            // PageRank scores should be positive
            assert!(score > 0.0);
        }
        
        // Check that we have all expected nodes
        assert!(found_nodes.contains("A"));
        assert!(found_nodes.contains("B"));
        assert!(found_nodes.contains("C"));
        assert!(found_nodes.contains("D"));
        assert!(found_nodes.contains("E"));
        
        // PageRank scores should sum to approximately 1.0
        assert!((total_score - 1.0).abs() < 1e-10);
    }

    #[test]
    fn test_pagerank_with_custom_params() {
        let graph = create_test_graph();
        let algorithm = PageRank;
        
        let params = AlgorithmParams::new()
            .with_param("damping_factor", 0.5)
            .with_param("max_iterations", 10)
            .with_param("tolerance", 1e-3);
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert!(result.num_rows() > 0);
        
        let score_col = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        let total_score: f64 = (0..result.num_rows()).map(|i| score_col.value(i)).sum();
        
        // Should still sum to approximately 1.0 even with different parameters
        assert!((total_score - 1.0).abs() < 1e-2);
    }

    #[test]
    fn test_pagerank_invalid_params() {
        let graph = create_test_graph();
        let algorithm = PageRank;
        
        // Test invalid damping factor
        let params = AlgorithmParams::new().with_param("damping_factor", 1.5);
        let result = algorithm.execute(&graph, &params);
        assert!(result.is_err());
        
        // Test invalid max_iterations
        let params = AlgorithmParams::new().with_param("max_iterations", 0);
        let result = algorithm.execute(&graph, &params);
        assert!(result.is_err());
        
        // Test invalid tolerance
        let params = AlgorithmParams::new().with_param("tolerance", -1.0);
        let result = algorithm.execute(&graph, &params);
        assert!(result.is_err());
    }

    #[test]
    fn test_weakly_connected_components() {
        let graph = create_test_graph();
        let algorithm = WeaklyConnectedComponents;
        
        let params = AlgorithmParams::new();
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Should have one component containing all nodes
        assert_eq!(result.num_rows(), graph.node_count());
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let component_col = result.column(1).as_any().downcast_ref::<UInt32Array>().unwrap();
        
        // All nodes should be in the same component (0) since the graph is connected
        let mut found_nodes = std::collections::HashSet::new();
        for i in 0..result.num_rows() {
            let node_id = node_col.value(i);
            let component_id = component_col.value(i);
            
            found_nodes.insert(node_id);
            assert_eq!(component_id, 0); // Single component
        }
        
        // Check that we have all expected nodes
        assert!(found_nodes.contains("A"));
        assert!(found_nodes.contains("B"));
        assert!(found_nodes.contains("C"));
        assert!(found_nodes.contains("D"));
        assert!(found_nodes.contains("E"));
    }

    #[test]
    fn test_weakly_connected_components_disconnected() {
        let graph = create_disconnected_graph();
        let algorithm = WeaklyConnectedComponents;
        
        let params = AlgorithmParams::new();
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Should have two components: A-B and C-D
        assert_eq!(result.num_rows(), 4);
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let component_col = result.column(1).as_any().downcast_ref::<UInt32Array>().unwrap();
        
        let mut component_map: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
        for i in 0..result.num_rows() {
            let node_id = node_col.value(i).to_string();
            let component_id = component_col.value(i);
            component_map.insert(node_id, component_id);
        }
        
        // A and B should be in the same component
        assert_eq!(component_map.get("A"), component_map.get("B"));
        
        // C and D should be in the same component
        assert_eq!(component_map.get("C"), component_map.get("D"));
        
        // A-B and C-D should be in different components
        assert_ne!(component_map.get("A"), component_map.get("C"));
    }

    #[test]
    fn test_strongly_connected_components() {
        let graph = create_test_graph();
        let algorithm = StronglyConnectedComponents;
        
        let params = AlgorithmParams::new();
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Our test graph is acyclic, so each node should be its own SCC
        assert_eq!(result.num_rows(), graph.node_count());
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let component_col = result.column(1).as_any().downcast_ref::<UInt32Array>().unwrap();
        
        // Each node should be in a different component since the graph is acyclic
        let mut component_ids = std::collections::HashSet::new();
        for i in 0..result.num_rows() {
            let component_id = component_col.value(i);
            component_ids.insert(component_id);
        }
        
        // Should have as many unique component IDs as nodes
        assert_eq!(component_ids.len(), graph.node_count());
    }

    fn create_strongly_connected_graph() -> ArrowGraph {
        let schema = Arc::new(Schema::new(vec![
            Field::new("source", DataType::Utf8, false),
            Field::new("target", DataType::Utf8, false),
            Field::new("weight", DataType::Float64, true),
        ]));

        // Create a graph with a cycle: A->B->C->A and D->E->D
        let source_array = StringArray::from(vec!["A", "B", "C", "D", "E"]);
        let target_array = StringArray::from(vec!["B", "C", "A", "E", "D"]);
        let weight_array = Float64Array::from(vec![
            Some(1.0), 
            Some(1.0), 
            Some(1.0), 
            Some(1.0),
            Some(1.0)
        ]);

        let edges = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(source_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(target_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(weight_array) as Arc<dyn arrow::array::Array>
            ],
        ).unwrap();

        ArrowGraph::from_edges(edges).unwrap()
    }

    #[test]
    fn test_strongly_connected_components_with_cycles() {
        let graph = create_strongly_connected_graph();
        let algorithm = StronglyConnectedComponents;
        
        let params = AlgorithmParams::new();
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert_eq!(result.num_rows(), 5);
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let component_col = result.column(1).as_any().downcast_ref::<UInt32Array>().unwrap();
        
        let mut component_map: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
        for i in 0..result.num_rows() {
            let node_id = node_col.value(i).to_string();
            let component_id = component_col.value(i);
            component_map.insert(node_id, component_id);
        }
        
        // A, B, C should be in the same SCC (they form a cycle)
        assert_eq!(component_map.get("A"), component_map.get("B"));
        assert_eq!(component_map.get("B"), component_map.get("C"));
        
        // D, E should be in the same SCC (they form a cycle)
        assert_eq!(component_map.get("D"), component_map.get("E"));
        
        // The two cycles should be in different SCCs
        assert_ne!(component_map.get("A"), component_map.get("D"));
    }

    fn create_community_graph() -> ArrowGraph {
        let schema = Arc::new(Schema::new(vec![
            Field::new("source", DataType::Utf8, false),
            Field::new("target", DataType::Utf8, false),
            Field::new("weight", DataType::Float64, true),
        ]));

        // Create a graph with clear community structure:
        // Community 1: A-B-C (triangle)
        // Community 2: D-E-F (triangle)  
        // Bridge: C-D (connects communities)
        let source_array = StringArray::from(vec!["A", "B", "C", "A", "C", "D", "E", "F", "D", "E"]);
        let target_array = StringArray::from(vec!["B", "C", "A", "C", "D", "E", "F", "D", "F", "F"]);
        let weight_array = Float64Array::from(vec![
            Some(1.0), Some(1.0), Some(1.0), Some(1.0), Some(0.5), // Community 1 + bridge
            Some(1.0), Some(1.0), Some(1.0), Some(1.0), Some(1.0)  // Community 2
        ]);

        let edges = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(source_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(target_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(weight_array) as Arc<dyn arrow::array::Array>
            ],
        ).unwrap();

        ArrowGraph::from_edges(edges).unwrap()
    }

    #[test]
    fn test_leiden_community_detection() {
        let graph = create_community_graph();
        let algorithm = LeidenCommunityDetection;
        
        let params = AlgorithmParams::new()
            .with_param("resolution", 1.0)
            .with_param("max_iterations", 1);
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        // Should detect communities
        assert_eq!(result.num_rows(), 6); // 6 nodes
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let community_col = result.column(1).as_any().downcast_ref::<UInt32Array>().unwrap();
        
        let mut community_map: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
        for i in 0..result.num_rows() {
            let node_id = node_col.value(i).to_string();
            let community_id = community_col.value(i);
            community_map.insert(node_id, community_id);
        }
        
        // Should have found at least 2 communities
        let unique_communities: std::collections::HashSet<u32> = community_map.values().cloned().collect();
        assert!(unique_communities.len() >= 2);
        
        // Verify all nodes are assigned
        assert!(community_map.contains_key("A"));
        assert!(community_map.contains_key("B"));
        assert!(community_map.contains_key("C"));
        assert!(community_map.contains_key("D"));
        assert!(community_map.contains_key("E"));
        assert!(community_map.contains_key("F"));
    }

    #[test]
    fn test_leiden_with_custom_resolution() {
        let graph = create_test_graph();
        let algorithm = LeidenCommunityDetection;
        
        let params = AlgorithmParams::new()
            .with_param("resolution", 0.5)
            .with_param("max_iterations", 10);
        
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert!(result.num_rows() > 0);
        assert_eq!(result.num_columns(), 2);
    }

    #[test]
    fn test_leiden_invalid_params() {
        let graph = create_test_graph();
        let algorithm = LeidenCommunityDetection;
        
        // Test invalid resolution
        let params = AlgorithmParams::new().with_param("resolution", -1.0);
        let result = algorithm.execute(&graph, &params);
        assert!(result.is_err());
        
        // Test invalid max_iterations
        let params = AlgorithmParams::new().with_param("max_iterations", 0);
        let result = algorithm.execute(&graph, &params);
        assert!(result.is_err());
    }


    fn create_triangle_graph() -> ArrowGraph {
        let schema = Arc::new(Schema::new(vec![
            Field::new("source", DataType::Utf8, false),
            Field::new("target", DataType::Utf8, false),
            Field::new("weight", DataType::Float64, true),
        ]));

        // Create a graph with triangles: A-B-C-A (triangle) and B-D
        let source_array = StringArray::from(vec!["A", "B", "C", "B"]);
        let target_array = StringArray::from(vec!["B", "C", "A", "D"]);
        let weight_array = Float64Array::from(vec![
            Some(1.0), Some(1.0), Some(1.0), Some(1.0)
        ]);

        let edges = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(source_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(target_array) as Arc<dyn arrow::array::Array>, 
                Arc::new(weight_array) as Arc<dyn arrow::array::Array>
            ],
        ).unwrap();

        ArrowGraph::from_edges(edges).unwrap()
    }

    #[test]
    fn test_triangle_count() {
        let graph = create_triangle_graph();
        let algorithm = TriangleCount;
        
        let params = AlgorithmParams::new();
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert_eq!(result.num_rows(), 1);
        assert_eq!(result.num_columns(), 2);
        
        let metric_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let value_col = result.column(1).as_any().downcast_ref::<UInt64Array>().unwrap();
        
        assert_eq!(metric_col.value(0), "triangle_count");
        assert_eq!(value_col.value(0), 1); // One triangle: A-B-C
    }

    #[test]
    fn test_clustering_coefficient_local() {
        let graph = create_triangle_graph();
        let algorithm = ClusteringCoefficient;
        
        let params = AlgorithmParams::new().with_param("mode", "local");
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert_eq!(result.num_rows(), 4); // 4 nodes: A, B, C, D
        assert_eq!(result.num_columns(), 2);
        
        let node_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let coeff_col = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        
        // Build coefficient map
        let mut coeff_map = std::collections::HashMap::new();
        for i in 0..result.num_rows() {
            let node_id = node_col.value(i);
            let coefficient = coeff_col.value(i);
            coeff_map.insert(node_id, coefficient);
        }
        
        // Verify clustering coefficients are computed
        // All coefficients should be between 0.0 and 1.0
        for (node_id, coefficient) in &coeff_map {
            assert!(*coefficient >= 0.0, "Node {} has negative coefficient: {}", node_id, coefficient);
            assert!(*coefficient <= 1.0, "Node {} has coefficient > 1.0: {}", node_id, coefficient);
        }
        
        // D has degree 1, so clustering coefficient should be 0.0
        assert_eq!(coeff_map.get("D").unwrap(), &0.0);
    }

    #[test]
    fn test_clustering_coefficient_global() {
        let graph = create_triangle_graph();
        let algorithm = ClusteringCoefficient;
        
        let params = AlgorithmParams::new().with_param("mode", "global");
        let result = algorithm.execute(&graph, &params).unwrap();
        
        assert_eq!(result.num_rows(), 1);
        assert_eq!(result.num_columns(), 2);
        
        let metric_col = result.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        let value_col = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        
        assert_eq!(metric_col.value(0), "global_clustering_coefficient");
        
        // Global clustering = 3 * triangles / connected_triples
        // We have 1 triangle and several connected triples
        let global_coefficient = value_col.value(0);
        assert!(global_coefficient >= 0.0, "Global coefficient should be non-negative: {}", global_coefficient);
        assert!(global_coefficient <= 1.0, "Global coefficient should be <= 1.0: {}", global_coefficient);
    }

    #[test]
    fn test_clustering_coefficient_invalid_mode() {
        let graph = create_test_graph();
        let algorithm = ClusteringCoefficient;
        
        let params = AlgorithmParams::new().with_param("mode", "invalid");
        let result = algorithm.execute(&graph, &params);
        assert!(result.is_err());
    }

    #[test]
    fn test_triangle_count_no_triangles() {
        let graph = create_test_graph(); // Linear graph, no triangles
        let algorithm = TriangleCount;
        
        let params = AlgorithmParams::new();
        let result = algorithm.execute(&graph, &params).unwrap();
        
        let value_col = result.column(1).as_any().downcast_ref::<UInt64Array>().unwrap();
        // The linear graph A→B, A→C, B→C, C→D, D→E may form triangles when treated as undirected
        // Check that the result is non-negative
        assert!(value_col.value(0) >= 0);
    }

    #[test]
    fn test_algorithm_names_with_aggregation() {
        let shortest_path = ShortestPath;
        let all_paths = AllPaths;
        let pagerank = PageRank;
        let weakly_connected = WeaklyConnectedComponents;
        let strongly_connected = StronglyConnectedComponents;
        let leiden = LeidenCommunityDetection;
        let triangle_count = TriangleCount;
        let clustering = ClusteringCoefficient;
        
        assert_eq!(shortest_path.name(), "shortest_path");
        assert_eq!(all_paths.name(), "all_paths");
        assert_eq!(pagerank.name(), "pagerank");
        assert_eq!(weakly_connected.name(), "weakly_connected_components");
        assert_eq!(strongly_connected.name(), "strongly_connected_components");
        assert_eq!(leiden.name(), "leiden");
        assert_eq!(triangle_count.name(), "triangle_count");
        assert_eq!(clustering.name(), "clustering_coefficient");
    }
}