qudag-protocol 0.5.0

use qudag_protocol::{Coordinator, ProtocolConfig, ProtocolState};
use std::time::Duration;
use tokio::time::sleep;

#[tokio::test]
async fn test_network_partition_detection() {
    // Test protocol behavior when network partitions are detected
    let config1 = ProtocolConfig {
        network: qudag_protocol::config::NetworkConfig {
            port: 10001,
            max_peers: 10,
            connect_timeout: Duration::from_secs(5),
        },
        ..Default::default()
    };
    
    let config2 = ProtocolConfig {
        network: qudag_protocol::config::NetworkConfig {
            port: 10002,
            max_peers: 10,
            connect_timeout: Duration::from_secs(5),
        },
        ..Default::default()
    };
    
    let mut coordinator1 = Coordinator::new(config1).await.unwrap();
    let mut coordinator2 = Coordinator::new(config2).await.unwrap();
    
    // Start both coordinators
    coordinator1.start().await.unwrap();
    coordinator2.start().await.unwrap();
    
    // Allow time for potential connection
    sleep(Duration::from_millis(100)).await;
    
    // Both should be running independently
    assert_eq!(coordinator1.state().await, ProtocolState::Running);
    assert_eq!(coordinator2.state().await, ProtocolState::Running);
    
    // Test message broadcasting during partition
    let message1 = vec![1, 2, 3];
    let message2 = vec![4, 5, 6];
    
    let result1 = coordinator1.broadcast_message(message1.clone()).await;
    let result2 = coordinator2.broadcast_message(message2.clone()).await;
    
    assert!(result1.is_ok());
    assert!(result2.is_ok());
    
    // Clean up
    coordinator1.stop().await.unwrap();
    coordinator2.stop().await.unwrap();
}

#[tokio::test]
async fn test_message_ordering_during_partition() {
    // Test that message ordering is preserved during network partitions
    let config = ProtocolConfig::default();
    let mut coordinator = Coordinator::new(config).await.unwrap();
    
    coordinator.start().await.unwrap();
    
    // Send ordered messages
    let messages = vec![
        vec![1, 0, 0],
        vec![2, 0, 0],
        vec![3, 0, 0],
        vec![4, 0, 0],
        vec![5, 0, 0],
    ];
    
    for message in &messages {
        let result = coordinator.broadcast_message(message.clone()).await;
        assert!(result.is_ok());
        
        // Small delay to ensure ordering
        sleep(Duration::from_millis(10)).await;
    }
    
    // Verify DAG has all messages
    if let Some(dag) = coordinator.dag_manager() {
        for message in &messages {
            assert!(dag.contains_message(message));
        }
    }
    
    coordinator.stop().await.unwrap();
}

#[tokio::test]
async fn test_partition_recovery_and_sync() {
    // Test protocol recovery and state synchronization after partition healing
    let config1 = ProtocolConfig {
        network: qudag_protocol::config::NetworkConfig {
            port: 10003,
            max_peers: 10,
            connect_timeout: Duration::from_secs(5),
        },
        ..Default::default()
    };
    
    let config2 = ProtocolConfig {
        network: qudag_protocol::config::NetworkConfig {
            port: 10004,
            max_peers: 10,
            connect_timeout: Duration::from_secs(5),
        },
        ..Default::default()
    };
    
    let mut coordinator1 = Coordinator::new(config1).await.unwrap();
    let mut coordinator2 = Coordinator::new(config2).await.unwrap();
    
    // Phase 1: Both nodes start isolated
    coordinator1.start().await.unwrap();
    coordinator2.start().await.unwrap();
    
    // Phase 2: Each node processes different messages during partition
    let messages1 = vec![vec![1, 1, 1], vec![2, 2, 2]];
    let messages2 = vec![vec![3, 3, 3], vec![4, 4, 4]];
    
    for message in &messages1 {
        coordinator1.broadcast_message(message.clone()).await.unwrap();
    }
    
    for message in &messages2 {
        coordinator2.broadcast_message(message.clone()).await.unwrap();
    }
    
    // Phase 3: Simulate partition healing (network reconnection)
    // TODO: Implement actual network connection between nodes
    sleep(Duration::from_millis(200)).await;
    
    // Phase 4: Verify state synchronization
    // In a real implementation, nodes would sync their DAG states
    // For now, verify both nodes are still running
    assert_eq!(coordinator1.state().await, ProtocolState::Running);
    assert_eq!(coordinator2.state().await, ProtocolState::Running);
    
    coordinator1.stop().await.unwrap();
    coordinator2.stop().await.unwrap();
}

#[tokio::test]
async fn test_consensus_during_partition() {
    // Test consensus behavior during network partitions
    let config = ProtocolConfig::default();
    let mut coordinator = Coordinator::new(config).await.unwrap();
    
    coordinator.start().await.unwrap();
    
    // Send conflicting messages that would require consensus
    let conflicting_messages = vec![
        vec![100, 1, 1],  // Base message
        vec![100, 1, 2],  // Potential conflict (same prefix)
        vec![100, 1, 3],  // Another potential conflict
    ];
    
    for message in &conflicting_messages {
        let result = coordinator.broadcast_message(message.clone()).await;
        assert!(result.is_ok());
        
        // Allow time for consensus processing
        sleep(Duration::from_millis(50)).await;
    }
    
    // Verify all messages are handled (consensus should resolve conflicts)
    if let Some(dag) = coordinator.dag_manager() {
        for message in &conflicting_messages {
            assert!(dag.contains_message(message));
        }
    }
    
    coordinator.stop().await.unwrap();
}

#[tokio::test]
async fn test_byzantine_behavior_during_partition() {
    // Test protocol resilience to Byzantine behavior during partitions
    let config1 = ProtocolConfig {
        network: qudag_protocol::config::NetworkConfig {
            port: 10005,
            max_peers: 10,
            connect_timeout: Duration::from_secs(5),
        },
        ..Default::default()
    };
    
    let config2 = ProtocolConfig {
        network: qudag_protocol::config::NetworkConfig {
            port: 10006,
            max_peers: 10,
            connect_timeout: Duration::from_secs(5),
        },
        ..Default::default()
    };
    
    let mut honest_coordinator = Coordinator::new(config1).await.unwrap();
    let mut byzantine_coordinator = Coordinator::new(config2).await.unwrap();
    
    honest_coordinator.start().await.unwrap();
    byzantine_coordinator.start().await.unwrap();
    
    // Honest node sends normal messages
    let honest_messages = vec![
        vec![200, 1],
        vec![200, 2],
        vec![200, 3],
    ];
    
    for message in &honest_messages {
        honest_coordinator.broadcast_message(message.clone()).await.unwrap();
    }
    
    // Byzantine node sends malicious/invalid messages
    let byzantine_messages = vec![
        vec![255; 1000], // Large invalid message
        vec![],          // Empty message
        vec![200, 1],    // Duplicate message
    ];
    
    for message in &byzantine_messages {
        let result = byzantine_coordinator.broadcast_message(message.clone()).await;
        // Byzantine messages should be handled gracefully
        assert!(result.is_ok());
    }
    
    // Honest node should remain functional
    assert_eq!(honest_coordinator.state().await, ProtocolState::Running);
    
    // Byzantine node should also remain stable (protocol should be resilient)
    assert_eq!(byzantine_coordinator.state().await, ProtocolState::Running);
    
    honest_coordinator.stop().await.unwrap();
    byzantine_coordinator.stop().await.unwrap();
}

#[tokio::test]
async fn test_dynamic_network_topology_changes() {
    // Test protocol adaptation to dynamic network topology changes
    let configs: Vec<ProtocolConfig> = (0..3).map(|i| {
        ProtocolConfig {
            network: qudag_protocol::config::NetworkConfig {
                port: 10007 + i as u16,
                max_peers: 10,
                connect_timeout: Duration::from_secs(5),
            },
            ..Default::default()
        }
    }).collect();
    
    let mut coordinators: Vec<Coordinator> = Vec::new();
    
    // Start all coordinators
    for config in configs {
        let mut coordinator = Coordinator::new(config).await.unwrap();
        coordinator.start().await.unwrap();
        coordinators.push(coordinator);
    }
    
    // Phase 1: All nodes send messages
    for (i, coordinator) in coordinators.iter_mut().enumerate() {
        let message = vec![i as u8, 100, 100];
        coordinator.broadcast_message(message).await.unwrap();
    }
    
    sleep(Duration::from_millis(100)).await;
    
    // Phase 2: Simulate topology change (one node leaves)
    coordinators[0].stop().await.unwrap();
    
    // Phase 3: Remaining nodes continue operation
    for (i, coordinator) in coordinators.iter_mut().enumerate().skip(1) {
        let message = vec![i as u8, 200, 200];
        let result = coordinator.broadcast_message(message).await;
        assert!(result.is_ok());
    }
    
    // Phase 4: Node rejoins network
    coordinators[0].start().await.unwrap();
    
    // Phase 5: All nodes send final messages
    for (i, coordinator) in coordinators.iter_mut().enumerate() {
        let message = vec![i as u8, 300, 300];
        let result = coordinator.broadcast_message(message).await;
        assert!(result.is_ok());
    }
    
    // Clean up
    for coordinator in coordinators.iter_mut() {
        coordinator.stop().await.unwrap();
    }
}

#[tokio::test]
async fn test_partition_tolerance_threshold() {
    // Test protocol behavior under different partition sizes
    let num_nodes = 5;
    let configs: Vec<ProtocolConfig> = (0..num_nodes).map(|i| {
        ProtocolConfig {
            network: qudag_protocol::config::NetworkConfig {
                port: 10010 + i as u16,
                max_peers: 10,
                connect_timeout: Duration::from_secs(5),
            },
            ..Default::default()
        }
    }).collect();
    
    let mut coordinators: Vec<Coordinator> = Vec::new();
    
    // Start all coordinators
    for config in configs {
        let mut coordinator = Coordinator::new(config).await.unwrap();
        coordinator.start().await.unwrap();
        coordinators.push(coordinator);
    }
    
    // Test with majority partition (3 out of 5 nodes)
    let majority_nodes = 3;
    
    // Majority partition sends messages
    for i in 0..majority_nodes {
        let message = vec![i as u8, 250, 250];
        coordinators[i].broadcast_message(message).await.unwrap();
    }
    
    // Minority partition sends messages
    for i in majority_nodes..num_nodes {
        let message = vec![i as u8, 251, 251];
        coordinators[i].broadcast_message(message).await.unwrap();
    }
    
    sleep(Duration::from_millis(200)).await;
    
    // All nodes should remain operational
    for coordinator in &coordinators {
        assert_eq!(coordinator.state().await, ProtocolState::Running);
    }
    
    // Clean up
    for coordinator in coordinators.iter_mut() {
        coordinator.stop().await.unwrap();
    }
}