use super::*;

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn two_process_cluster_exposes_control_plane_on_both_nodes() {
    if std::env::var("CARGO_BIN_EXE_kafkalite").is_err() {
        return;
    }
    let tempdir = tempdir().unwrap();
    let broker1 = free_port();
    let broker2 = free_port();
    let controller1 = free_port();
    let controller2 = free_port();
    let quorum = format!("1@127.0.0.1:{controller1},2@127.0.0.1:{controller2}");

    let mut node1 = spawn_cluster_process(tempdir.path(), 1, broker1, controller1, &quorum);
    let mut node2 = spawn_cluster_process(tempdir.path(), 2, broker2, controller2, &quorum);
    wait_until_broker_ready(&node1.bootstrap, Duration::from_secs(10)).unwrap();
    wait_until_broker_ready(&node2.bootstrap, Duration::from_secs(10)).unwrap();

    let transport = TcpClusterRpcTransport;
    let response1 = transport
        .send_to(
            &node1.controller_target,
            ClusterRpcRequest::Vote(VoteRequest {
                term: 1,
                candidate_id: 1,
                last_metadata_offset: -1,
            }),
        )
        .await
        .unwrap();
    let response2 = transport
        .send_to(
            &node2.controller_target,
            ClusterRpcRequest::Vote(VoteRequest {
                term: 1,
                candidate_id: 2,
                last_metadata_offset: -1,
            }),
        )
        .await
        .unwrap();

    assert!(matches!(response1, ClusterRpcResponse::Vote(_)));
    assert!(matches!(response2, ClusterRpcResponse::Vote(_)));

    let _ = node1.child.kill();
    let _ = node1.child.wait();
    let _ = node2.child.kill();
    let _ = node2.child.wait();
}
