netdesk/native/core/tests/host_viewer.rs

//! End-to-end: the core HOST (host.rs) serves a session that the core VIEWER
//! (signaling.rs connect) authenticates to, over the LIVE server. Proves the
//! Android host's protocol (register -> challenge -> auth/E2E -> accept) and
//! input routing interoperate before any Kotlin is wired.
//!
//! Network test -> #[ignore] by default. Run explicitly:
//!   cargo test -p netdesk-core --target x86_64-pc-windows-gnu --test host_viewer -- --ignored --nocapture

use netdesk_core::crypto::{gen_host_id, new_host_credentials};
use netdesk_core::host::{run_host, HostConfig, HostEvent, HostFrame};
use netdesk_core::{connect, ConnectConfig, SessionEvent};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::timeout;

async fn wait_host<F: Fn(&HostEvent) -> bool>(
    rx: &mut mpsc::UnboundedReceiver<HostEvent>,
    pred: F,
    secs: u64,
) -> Option<HostEvent> {
    let deadline = Duration::from_secs(secs);
    loop {
        match timeout(deadline, rx.recv()).await {
            Ok(Some(ev)) => {
                if pred(&ev) {
                    return Some(ev);
                }
            }
            _ => return None,
        }
    }
}

async fn wait_view<F: Fn(&SessionEvent) -> bool>(
    rx: &mut mpsc::UnboundedReceiver<SessionEvent>,
    pred: F,
    secs: u64,
) -> Option<SessionEvent> {
    let deadline = Duration::from_secs(secs);
    loop {
        match timeout(deadline, rx.recv()).await {
            Ok(Some(ev)) => {
                if pred(&ev) {
                    return Some(ev);
                }
            }
            _ => return None,
        }
    }
}

#[tokio::test]
#[ignore]
async fn host_viewer_handshake_and_input() {
    let ws = "wss://netdesk.space/ws";
    let id = gen_host_id();
    let password = "e2e-host-test-pw";
    let (salt_b64, iters, verifier_hex) = new_host_credentials(password);

    let (frame_tx, frame_rx) = mpsc::unbounded_channel::<HostFrame>();
    let (hev_tx, mut hev_rx) = mpsc::unbounded_channel::<HostEvent>();
    let hcfg = HostConfig {
        ws_url: ws.into(),
        host_id: id.clone(),
        name: "TestAndroidHost".into(),
        salt_b64,
        iters,
        verifier_hex,
        width: 1080,
        height: 1920,
        audio: false,
        monitors: Vec::new(),
        monitor: 0,
        perms: Default::default(),
        approve: false,
        totp_secret: String::new(),
    };
    // Outbound control channel (unused by this test; drop the sender).
    let (_hctl_tx, hctl_rx) = tokio::sync::mpsc::unbounded_channel();
    let host_task =
        tokio::spawn(async move { run_host(hcfg, frame_rx, hev_tx, hctl_rx).await });

    assert!(
        wait_host(&mut hev_rx, |e| matches!(e, HostEvent::Registered), 10)
            .await
            .is_some(),
        "host did not register"
    );
    // let the registration settle on the server before the viewer dials in
    tokio::time::sleep(Duration::from_millis(600)).await;

    // viewer connects to our host id
    let (vtx, mut vrx) = mpsc::unbounded_channel::<SessionEvent>();
    let vcfg = ConnectConfig {
        ws_url: ws.into(),
        host_id: id.clone(),
        password: password.into(),
        totp: String::new(),
        takeover: false,
    };
    let session = connect(vcfg, vtx).await.expect("viewer connect+handshake");

    // host sees the viewer; viewer sees Connected with our advertised size
    assert!(
        wait_host(&mut hev_rx, |e| matches!(e, HostEvent::ViewerConnected { .. }), 10)
            .await
            .is_some(),
        "host never saw ViewerConnected"
    );
    let connected = wait_view(&mut vrx, |e| matches!(e, SessionEvent::Connected { .. }), 10).await;
    match connected {
        Some(SessionEvent::Connected { w, h, e2e, .. }) => {
            assert_eq!((w, h), (1080, 1920), "advertised resolution");
            assert!(e2e, "E2E negotiated");
        }
        _ => panic!("viewer never got Connected"),
    }

    // viewer input must surface on the host side with matching coordinates
    session.mouse(123, 456, 1, 0.0, 0.0);
    match wait_host(&mut hev_rx, |e| matches!(e, HostEvent::Mouse { .. }), 10).await {
        Some(HostEvent::Mouse { x, y, buttons, .. }) => {
            assert_eq!((x, y, buttons), (123, 456, 1), "mouse routed to host");
        }
        _ => panic!("host never received the viewer's mouse"),
    }

    session.key("KeyA", true);
    match wait_host(&mut hev_rx, |e| matches!(e, HostEvent::Key { .. }), 10).await {
        Some(HostEvent::Key { code, down }) => {
            assert_eq!((code.as_str(), down), ("KeyA", true), "key routed to host");
        }
        _ => panic!("host never received the viewer's key"),
    }

    // a frame fed to the host reaches the viewer as a binary message (we only
    // assert the host accepts + seals it; decode is covered elsewhere).
    let _ = frame_tx.send(HostFrame {
        data: vec![0, 0, 0, 1, 0x67, 0x42, 0, 0x28],
        keyframe: true,
        codec: 0,
    });
    tokio::time::sleep(Duration::from_millis(300)).await;

    session.close();
    drop(frame_tx);
    let _ = timeout(Duration::from_secs(3), host_task).await;
    println!("HOST_VIEWER_E2E_OK id={id}");
}