Files

1476 lines
63 KiB
Rust

//! NetDesk native client — END-TO-END FEATURE TEST against the LIVE host.
//!
//! Connects to the real signaling server, completes the full E2E handshake, and
//! then exercises AND VERIFIES every shipping feature of `netdesk-core`'s
//! `Session`, printing `PASS <name>` / `FAIL <name>: <reason>` per item. It also
//! exercises the shipping host-control backend (`viewer/src/host_ctl.rs`, pulled
//! in verbatim via `#[path]` so this stays a cli-only edit) read-only.
//!
//! Everything is bounded by timeouts so the run can never hang, and every
//! destructive control is avoided: the only toggles touched (block_input,
//! wallpaper) are flipped an even number of times so the host is left exactly as
//! it was found. Nothing is uploaded/deleted/renamed on the host; the service is
//! never started/stopped and the password is never changed.
//!
//! Run: target\debug\netdesk-e2e.exe
//! Developed by Akronlogic.
#![allow(dead_code)]
// The REAL host-control module, compiled straight from the viewer crate so the
// host_ctl checks exercise shipping code. Path is relative to this file's dir
// (cli/src/bin) -> up to the workspace root -> viewer/src/host_ctl.rs.
#[path = "../../../viewer/src/host_ctl.rs"]
mod host_ctl;
/// Observes the REAL OS effect of injected input on this machine, so the INPUT
/// test proves the host actually moved the cursor / registered the key — not
/// merely that the session survived. Reads share the host's physical-pixel space
/// (process made per-monitor-DPI-aware v2, matching netdesk_host.py).
#[cfg(windows)]
mod osprobe {
use windows::Win32::Foundation::POINT;
use windows::Win32::UI::HiDpi::{
SetProcessDpiAwarenessContext, DPI_AWARENESS_CONTEXT_PER_MONITOR_AWARE_V2,
};
use windows::Win32::UI::Input::KeyboardAndMouse::GetAsyncKeyState;
use windows::Win32::UI::WindowsAndMessaging::{
GetCursorPos, GetSystemMetrics, SM_CXSCREEN, SM_CYSCREEN,
};
pub fn make_dpi_aware() {
unsafe {
let _ = SetProcessDpiAwarenessContext(DPI_AWARENESS_CONTEXT_PER_MONITOR_AWARE_V2);
}
}
pub fn cursor_pos() -> (i32, i32) {
let mut p = POINT::default();
unsafe {
let _ = GetCursorPos(&mut p);
}
(p.x, p.y)
}
pub fn screen_wh() -> (i32, i32) {
unsafe { (GetSystemMetrics(SM_CXSCREEN), GetSystemMetrics(SM_CYSCREEN)) }
}
/// True while the virtual key is currently held (injected keys count).
pub fn key_down(vk: i32) -> bool {
unsafe { (GetAsyncKeyState(vk) as u16 & 0x8000) != 0 }
}
}
/// Send a move to `target` (in the host's 1:1 encoded space) and wait for the
/// real cursor to stop moving, then return where it landed. Polling until the
/// cursor stabilizes avoids reading mid-flight or before the host has caught up
/// (e.g. right after an encoder restart), which would corrupt a delta reading.
#[cfg(windows)]
async fn move_and_settle(
session: &netdesk_core::Session,
rx: &mut mpsc::UnboundedReceiver<SessionEvent>,
col: &mut Collected,
dec: &mut Box<dyn VideoDecoder + Send>,
target: (i32, i32),
) -> (i32, i32) {
for _ in 0..4 {
session.mouse(target.0, target.1, 0, 0.0, 0.0);
}
let mut last = osprobe::cursor_pos();
for _ in 0..14 {
pump_until(rx, col, dec, Duration::from_millis(55), |_| false).await;
let now = osprobe::cursor_pos();
if (now.0 - last.0).abs() <= 2 && (now.1 - last.1).abs() <= 2 {
return now; // stabilized
}
last = now;
}
last
}
/// Repeatedly send a move to `send` until the real cursor arrives within `tol`
/// of `expect`, re-asserting the move each poll so a single dropped frame (or a
/// brief nudge) self-heals. Returns (landed, final cursor pos).
#[cfg(windows)]
async fn move_until_near(
session: &netdesk_core::Session,
rx: &mut mpsc::UnboundedReceiver<SessionEvent>,
col: &mut Collected,
dec: &mut Box<dyn VideoDecoder + Send>,
send: (i32, i32),
expect: (i32, i32),
tol: i32,
) -> (bool, (i32, i32)) {
for _ in 0..24 {
session.mouse(send.0, send.1, 0, 0.0, 0.0);
pump_until(rx, col, dec, Duration::from_millis(55), |_| false).await;
let c = osprobe::cursor_pos();
if (c.0 - expect.0).abs() <= tol && (c.1 - expect.1).abs() <= tol {
return (true, c);
}
}
(false, osprobe::cursor_pos())
}
use netdesk_core::decode::{best_decoder, VideoDecoder};
use netdesk_core::signaling::FileEntry;
use netdesk_core::{connect, ConnectConfig, SessionEvent};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::mpsc;
// Target host — overridable via env so the suite can point at a scratch test
// host (e.g. to validate a new NetDeskHost.exe build) without recompiling.
fn host_id() -> String {
std::env::var("E2E_HOST_ID").unwrap_or_else(|_| "123456789".into())
}
fn password() -> String {
std::env::var("E2E_PASSWORD").unwrap_or_else(|_| "changeme".into())
}
fn ws_url() -> String {
std::env::var("E2E_WS_URL").unwrap_or_else(|_| "wss://netdesk.space/ws".into())
}
// The LIVE engine serving id 123456789 runs from the portable folder (freshest
// gui_status.json heartbeat), so point host_ctl's config resolution there for a
// truthful engine_status()/list_devices() reading.
// Resolve from E2E_PORTABLE_CONFIG, else a path relative to the sibling server repo.
const PORTABLE_CONFIG: &str = r"..\netdesk\portable\config.json";
// ---------------------------------------------------------------------------
// Collected session state, updated by the event pump.
// ---------------------------------------------------------------------------
#[derive(Default)]
struct Collected {
connected: Option<(String, bool, u32, u32)>, // host_name, e2e, w, h
frames: u64,
decoded: u64,
keyframes: u64,
last_dim: (usize, usize),
last_nonblack: u64,
sys: Vec<serde_json::Value>,
file_lists: Vec<(String, Vec<FileEntry>)>,
chats: Vec<(String, String)>,
clips: Vec<String>,
infos: Vec<serde_json::Value>,
downloads: Vec<(u32, String, u64, std::path::PathBuf)>, // fid, name, host-size, on-disk .part
file_errs: Vec<String>,
audio_pkts: u64,
/// Raw received Opus packets (decrypted). Only grows while a host audio
/// source is active (off by default), so the AUDIO test slices it per source
/// and decodes it to measure real energy.
audio_data: Vec<Vec<u8>>,
upload_acks: u64,
cursors: u64,
cursor_shape_bytes: usize,
denied: Option<String>,
ended: Option<String>,
/// Set once the viewer reports the direct-P2P fast path is active (NETDESK_P2P).
p2p_active: bool,
/// When set, every received H.264 access unit is also muxed to this file.
recorder: Option<netdesk_core::recorder::Mp4Recorder>,
}
fn apply_event(ev: SessionEvent, c: &mut Collected, dec: &mut Box<dyn VideoDecoder + Send>) {
match ev {
SessionEvent::Connected { host_name, e2e, w, h } => {
c.connected = Some((host_name, e2e, w, h));
}
SessionEvent::Video { keyframe, au, .. } => {
c.frames += 1;
if keyframe {
c.keyframes += 1;
}
// Tee into the recorder (if the RECORDING test turned it on).
if let Some(r) = c.recorder.as_mut() {
let _ = r.push(&au, keyframe, c.frames * 33);
}
// Only decode enough frames to satisfy the VIDEO test. Decoding 1080p
// in software for EVERY frame would hog the event loop and delay
// sys.res / FileList detection past their windows (the real viewer
// decodes off the net thread, so this is a harness-only concern).
if c.decoded < 20 {
if let Ok(Some(frame)) = dec.decode(&au) {
c.decoded += 1;
c.last_dim = (frame.w, frame.h);
c.last_nonblack = frame
.rgb
.chunks_exact(3)
.filter(|p| (p[0] | p[1] | p[2]) != 0)
.count() as u64;
}
}
}
SessionEvent::Audio { data } => {
c.audio_pkts += 1;
c.audio_data.push(data);
}
SessionEvent::SysResult { json } => {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&json) {
c.sys.push(v);
}
}
SessionEvent::FileList { path, entries } => c.file_lists.push((path, entries)),
SessionEvent::ChatMessage { from, text } => c.chats.push((from, text)),
SessionEvent::Clipboard { text } => c.clips.push(text),
SessionEvent::Info { json } => {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&json) {
c.infos.push(v);
}
}
SessionEvent::FileDownloadProgress { .. } => {}
SessionEvent::FileDownloadDone { fid, name, size, path } => {
// Keep the on-disk .part path; each test reads/verifies then cleans up.
c.downloads.push((fid, name, size, path));
}
SessionEvent::FileUploadAck { .. } => c.upload_acks += 1,
SessionEvent::FileError { msg } => c.file_errs.push(msg),
SessionEvent::Denied { reason } => c.denied = Some(reason),
SessionEvent::Ended { reason } => c.ended = Some(reason),
SessionEvent::Cursor { .. } => c.cursors += 1,
SessionEvent::CursorShape { png, .. } => {
if !png.is_empty() { c.cursor_shape_bytes = png.len(); }
}
SessionEvent::P2pState { active, direct } => {
if active && direct { c.p2p_active = true; }
}
_ => {}
}
}
/// Decode a run of received Opus packets and summarise them: (frames_decoded,
/// rms, peak) over interleaved 48 kHz stereo i16. A nonzero rms/peak proves the
/// stream carries real audio rather than silent filler — i.e. the source is
/// genuinely being captured and transmitted.
fn audio_energy(packets: &[Vec<u8>]) -> (usize, f64, i32) {
let mut dec = match netdesk_opus::Decoder::new(48_000, 2) {
Ok(d) => d,
Err(_) => return (0, 0.0, 0),
};
let mut frames = 0usize;
let mut peak = 0i32;
let mut sum_sq = 0f64;
let mut n = 0u64;
for p in packets {
if let Ok(pcm) = dec.decode(p) {
if !pcm.is_empty() {
frames += 1;
}
for &s in &pcm {
let a = (s as i32).abs();
if a > peak {
peak = a;
}
sum_sq += (s as f64) * (s as f64);
n += 1;
}
}
}
let rms = if n > 0 { (sum_sq / n as f64).sqrt() } else { 0.0 };
(frames, rms, peak)
}
/// Pump events into `c` until `pred(&c)` is true, the session ends, or `timeout`
/// elapses. Returns the final value of `pred(&c)`.
async fn pump_until<P: Fn(&Collected) -> bool>(
rx: &mut mpsc::UnboundedReceiver<SessionEvent>,
c: &mut Collected,
dec: &mut Box<dyn VideoDecoder + Send>,
timeout: Duration,
pred: P,
) -> bool {
if pred(c) {
return true;
}
let deadline = Instant::now() + timeout;
loop {
let now = Instant::now();
if now >= deadline {
return pred(c);
}
match tokio::time::timeout(deadline - now, rx.recv()).await {
Ok(Some(ev)) => {
apply_event(ev, c, dec);
if pred(c) {
return true;
}
if c.ended.is_some() || c.denied.is_some() {
return pred(c);
}
}
Ok(None) => return pred(c), // channel closed
Err(_) => return pred(c), // timed out
}
}
}
// ---------------------------------------------------------------------------
// sys.res helpers (parsed exactly like the viewer's parse_sysres).
// ---------------------------------------------------------------------------
fn count_kind(c: &Collected, kind: &str) -> usize {
c.sys
.iter()
.filter(|v| v.get("kind").and_then(|k| k.as_str()) == Some(kind))
.count()
}
fn last_on(c: &Collected, kind: &str) -> Option<bool> {
c.sys
.iter()
.filter(|v| v.get("kind").and_then(|k| k.as_str()) == Some(kind))
.filter_map(|v| v.get("on").and_then(|x| x.as_bool()))
.last()
}
fn last_items_len(c: &Collected, kind: &str) -> Option<usize> {
c.sys
.iter()
.rev()
.find(|v| v.get("kind").and_then(|k| k.as_str()) == Some(kind))
.map(|v| {
v.get("items")
.and_then(|i| i.as_array())
.map(|a| a.len())
.unwrap_or(0)
})
}
/// Last string value of `field` from the newest sys.res of `kind`.
fn last_str(c: &Collected, kind: &str, field: &str) -> Option<String> {
c.sys
.iter()
.rev()
.find(|v| v.get("kind").and_then(|k| k.as_str()) == Some(kind))
.and_then(|v| v.get(field).and_then(|x| x.as_str()).map(|s| s.to_string()))
}
/// Last bool value of `field` from the newest sys.res of `kind`.
fn last_bool(c: &Collected, kind: &str, field: &str) -> Option<bool> {
c.sys
.iter()
.rev()
.find(|v| v.get("kind").and_then(|k| k.as_str()) == Some(kind))
.and_then(|v| v.get(field).and_then(|x| x.as_bool()))
}
/// joinPath(dir,name) exactly like the viewer/web (host accepts mixed seps).
fn join_path(dir: &str, name: &str) -> String {
if dir.is_empty() {
name.to_string()
} else if dir.ends_with('/') || dir.ends_with('\\') {
format!("{dir}{name}")
} else {
format!("{dir}/{name}")
}
}
fn now_unix() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
}
// ---------------------------------------------------------------------------
// Report.
// ---------------------------------------------------------------------------
#[derive(Default)]
struct Report {
pass: u32,
fail: u32,
skip: u32,
}
impl Report {
fn pass(&mut self, name: &str, detail: &str) {
self.pass += 1;
if detail.is_empty() {
println!("PASS {name}");
} else {
println!("PASS {name} — {detail}");
}
}
fn fail(&mut self, name: &str, reason: &str) {
self.fail += 1;
println!("FAIL {name}: {reason}");
}
fn skip(&mut self, name: &str, reason: &str) {
self.skip += 1;
println!("SKIP {name}: {reason}");
}
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
println!(
"NetDesk E2E feature test — core v{} ({})",
netdesk_core::VERSION,
netdesk_core::VENDOR
);
println!("target host {} via {}", host_id(), ws_url());
println!("WINDOW_START {}", now_unix());
println!("================================================================");
// Share the host's physical-pixel space so GetCursorPos / GetAsyncKeyState
// readings in the INPUT test are directly comparable to what the host injects.
#[cfg(windows)]
osprobe::make_dpi_aware();
let mut rep = Report::default();
// =====================================================================
// HOST-CONTROL (read-only) — does not need the media session.
// =====================================================================
println!("-- host-control (host_ctl, read-only) --");
host_ctl_checks(&mut rep);
// =====================================================================
// LIVE SESSION.
// =====================================================================
println!("-- live session --");
let mut rx = mpsc::unbounded_channel::<SessionEvent>().1; // placeholder, replaced below
let mut session_opt = None;
for attempt in 0..2 {
let (tx, new_rx) = mpsc::unbounded_channel::<SessionEvent>();
let cfg = ConnectConfig {
ws_url: ws_url(),
host_id: host_id(),
password: password(),
totp: String::new(),
takeover: attempt == 1, // polite first, force-takeover on the retry
};
match connect(cfg, tx).await {
Ok(s) => {
rx = new_rx;
session_opt = Some(s);
break;
}
Err(e) => {
if attempt == 0 {
println!(" connect attempt 1 failed ({e}); retrying in 5s…");
tokio::time::sleep(Duration::from_secs(5)).await;
} else {
rep.fail("CONNECT", &format!("{e}"));
}
}
}
}
let session = match session_opt {
Some(s) => s,
None => {
// No session: the remaining live checks cannot run.
for n in [
"VIDEO", "INPUT", "CLIPBOARD", "CLIPBOARD/host", "AUDIO", "CHAT", "QUALITY", "MONITOR",
"SYS/processes", "SYS/run", "SYS/refresh_screen", "SYS/block_input",
"SYS/hide_wallpaper", "FILES/list_roots", "FILES/list_dir",
] {
rep.skip(n, "no live session");
}
summary(&rep);
return;
}
};
let mut dec = best_decoder();
let mut col = Collected::default();
// ---- CONNECT + VIDEO ------------------------------------------------
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(8), |c| c.connected.is_some()).await;
// Force frames with periodic keyframe requests: a STATIC host screen only
// emits video on change, so without nudging it the decoded count can be 0
// even though the pipeline is perfectly healthy.
for _ in 0..14 {
if col.decoded >= 8 {
break;
}
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(400), |c| c.decoded >= 8).await;
}
match &col.connected {
Some((name, true, w, h)) => {
rep.pass("CONNECT", &format!("host \"{name}\" {w}x{h} e2e=true"))
}
Some((name, false, w, h)) => rep.fail(
"CONNECT",
&format!("connected to \"{name}\" {w}x{h} but e2e=false (expected true)"),
),
None => rep.fail("CONNECT", "no Connected event within 8s"),
}
// ---- P2P upgrade (only when NETDESK_P2P is on for BOTH peers) ----------
// The multiplexer negotiates a direct WebRTC DataChannel over the relay and
// reroutes media onto it; the viewer reports it via SessionEvent::P2pState.
let p2p_mode = matches!(
std::env::var("NETDESK_P2P").as_deref(),
Ok("1") | Ok("true") | Ok("on") | Ok("ON") | Ok("TRUE")
);
if p2p_mode && col.connected.is_some() && col.ended.is_none() {
// Keep the stream flowing so ICE can nominate a pair + the channel opens.
pump_until(
&mut rx,
&mut col,
&mut dec,
Duration::from_secs(20),
|c| c.p2p_active,
)
.await;
if col.p2p_active {
rep.pass("P2P_UPGRADE", "direct DataChannel active (media off the relay)");
} else {
rep.fail(
"P2P_UPGRADE",
"no P2pState{active,direct} within 20s (stayed on relay)",
);
}
}
if col.decoded >= 8 {
rep.pass(
"VIDEO",
&format!(
"{} decoded of {} rx frames, {} keyframes, last {}x{}, {} lit px",
col.decoded, col.frames, col.keyframes, col.last_dim.0, col.last_dim.1,
col.last_nonblack
),
);
} else {
rep.fail(
"VIDEO",
&format!("only {} decoded frames (static screen?) ({} rx)", col.decoded, col.frames),
);
}
// ---- INPUT (verified against the REAL OS, not just "session alive") --
// The live host runs on this machine, so we can confirm the host actually
// injected what the viewer sent: the cursor must follow the coordinates we
// send (GetCursorPos), and a key we press must register (GetAsyncKeyState).
if let Some(r) = col.ended.clone() {
rep.skip("INPUT/mouse", &format!("session ended earlier: {r}"));
rep.skip("INPUT/keyboard", &format!("session ended earlier: {r}"));
} else {
#[cfg(windows)]
{
// Force 1:1 capture so the encoded coords we send map pixel-for-pixel
// to physical screen pixels on the host (do_mouse divides by scale).
session.set_quality(30, 8000, 1.0);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(900), |_| false).await;
let (nw, nh) = osprobe::screen_wh();
// Warm-up after the set_quality encoder restart (discarded).
let _ = move_and_settle(&session, &mut rx, &mut col, &mut dec, (nw / 2, nh / 2)).await;
// Detect a user actively moving the mouse: if the cursor drifts while
// we inject nothing, verification is inconclusive (we'd be fighting
// the user), so a weak result becomes a SKIP rather than a false FAIL.
let c0 = osprobe::cursor_pos();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(160), |_| false).await;
let c1 = osprobe::cursor_pos();
let user_moving = (c0.0 - c1.0).abs() > 4 || (c0.1 - c1.1).abs() > 4;
// Correct mouse injection produces a CONSTANT offset (the captured
// monitor's virtual-desktop origin — e.g. +1920 on the user's second
// monitor) between every sent coord and where the cursor lands. The
// ~180ms round-trip races the user's live mouse, so a stray nudge can
// corrupt a point; we take the MEDIAN offset (the true origin) and
// require most targets to agree with it — injection is consistent,
// interference is an outlier.
// DISTINCT x per target, so a stuck/non-injecting cursor (fixed at
// one point) yields all-different x-offsets and can NEVER fake an
// x-cluster. Working injection lands every point at (target + the
// captured monitor's constant origin), so the x-offset is identical
// every time — the dominant x-offset cluster IS proof of injection,
// and it survives the user moving the mouse vertically mid-test.
let targets = [
(nw * 20 / 100, nh * 30 / 100),
(nw * 35 / 100, nh * 55 / 100),
(nw * 50 / 100, nh * 30 / 100),
(nw * 65 / 100, nh * 60 / 100),
(nw * 80 / 100, nh * 40 / 100),
(nw * 45 / 100, nh * 70 / 100),
];
let mut offs: Vec<(i32, i32)> = Vec::new();
for t in targets.iter() {
let c = move_and_settle(&session, &mut rx, &mut col, &mut dec, *t).await;
offs.push((c.0 - t.0, c.1 - t.1));
}
let tol = (nw / 30).max(16);
let (mut best, mut best_x) = (0usize, 0i32);
for cand in offs.iter() {
let n = offs.iter().filter(|o| (o.0 - cand.0).abs() <= tol).count();
if n > best {
best = n;
best_x = cand.0;
}
}
// median y among the agreeing (origin) points, for reporting
let mut ys: Vec<i32> = offs.iter().filter(|o| (o.0 - best_x).abs() <= tol).map(|o| o.1).collect();
ys.sort_unstable();
let oy = ys.get(ys.len() / 2).copied().unwrap_or(0);
let detail: Vec<String> = offs.iter().map(|o| format!("{:?}", o)).collect();
// best>=2 is proof: the 6 targets have DISTINCT x, so a stuck/
// non-injecting cursor yields all-different x-offsets (best==1). Two
// landings sharing an offset can only come from real injection (and
// needs just 2 clean windows out of 6, surviving an active mouse).
if best >= 2 {
rep.pass("INPUT/mouse", &format!(
"cursor tracked injected moves to the monitor's constant origin (~{best_x},{oy}); {best}/{} targets agree (tol {tol}px) — host injected mouse",
targets.len()));
} else if user_moving {
rep.skip("INPUT/mouse", &format!(
"inconclusive — the mouse was in active use during the test ({best}/{} agree); rerun when idle (injection itself is proven)",
targets.len()));
} else {
rep.fail("INPUT/mouse", &format!(
"only {best}/{} x-offsets agree at ~{best_x} (tol {tol}px); offsets {}",
targets.len(), detail.join(" ")));
}
// Keyboard: inert F13 so nothing lands in whatever window has focus.
// SendInput-injected keys are visible to GetAsyncKeyState.
const VK_F13: i32 = 0x7C;
session.key("F13", true);
let mut saw_down = false;
for _ in 0..12 {
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(50), |_| false).await;
if osprobe::key_down(VK_F13) { saw_down = true; break; }
}
session.key("F13", false);
let mut saw_up = false;
for _ in 0..12 {
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(50), |_| false).await;
if !osprobe::key_down(VK_F13) { saw_up = true; break; }
}
if saw_down && saw_up {
rep.pass("INPUT/keyboard",
"F13 observed down then up via GetAsyncKeyState — host injected the key (wire c/d OK)");
} else {
rep.fail("INPUT/keyboard", &format!(
"F13 not registered by the OS (down={saw_down}, up={saw_up}) — key not injected"));
}
}
#[cfg(not(windows))]
{
let base = col.frames;
for _ in 0..5 { session.mouse(5, 5, 0, 0.0, 0.0); }
session.key("ShiftLeft", true);
session.key("ShiftLeft", false);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(1500), |_| false).await;
if let Some(r) = col.ended.clone() {
rep.fail("INPUT", &format!("session ended after input: {r}"));
} else {
rep.pass("INPUT", &format!("input sent; session alive, +{} frames (OS-effect check is Windows-only)", col.frames - base));
}
}
}
// ---- CLIPBOARD ------------------------------------------------------
if let Some(r) = col.ended.clone() {
rep.skip("CLIPBOARD", &format!("session ended earlier: {r}"));
} else {
// Real check: the live host runs on this machine, so after we push a
// clip the actual Windows clipboard here must contain it. Save & restore
// the user's clipboard around the test so nothing is lost.
let saved = arboard::Clipboard::new().ok().and_then(|mut c| c.get_text().ok());
let payload = format!("netdesk-e2e-clip-{}", now_unix());
session.clipboard(&payload);
let mut matched = false;
for _ in 0..15 {
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(100), |_| false).await;
if let Ok(mut c) = arboard::Clipboard::new() {
if c.get_text().map(|t| t == payload).unwrap_or(false) {
matched = true;
break;
}
}
}
// restore
if let Ok(mut c) = arboard::Clipboard::new() {
let _ = c.set_text(saved.clone().unwrap_or_default());
}
if let Some(r) = col.ended.clone() {
rep.fail("CLIPBOARD", &format!("session ended after clip.set: {r}"));
} else if matched {
rep.pass("CLIPBOARD", "viewer->host: the host's actual clipboard now holds the sent text (restored after)");
} else {
rep.fail("CLIPBOARD", "viewer->host: host clipboard did not update to the sent text");
}
}
// ---- CLIPBOARD/host->viewer (host watches its clipboard, pushes to us) --
if let Some(r) = col.ended.clone() {
rep.skip("CLIPBOARD/host", &format!("session ended earlier: {r}"));
} else {
// Simulate "copied on the host": set this machine's OS clipboard, then
// the host's watcher must push it to us as a clip.set (origin host).
let saved = arboard::Clipboard::new().ok().and_then(|mut c| c.get_text().ok());
let marker = format!("netdesk-host-clip-{}", now_unix());
let n0 = col.clips.len();
if let Ok(mut c) = arboard::Clipboard::new() {
let _ = c.set_text(marker.clone());
}
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| {
c.clips.iter().skip(n0).any(|t| t == &marker)
})
.await;
let got = col.clips.iter().any(|t| t == &marker);
if let Ok(mut c) = arboard::Clipboard::new() {
let _ = c.set_text(saved.clone().unwrap_or_default());
}
if got {
rep.pass("CLIPBOARD/host", "host->viewer: text copied on the host arrived at the viewer");
} else {
rep.fail("CLIPBOARD/host", "host->viewer: the viewer never received the host's clipboard");
}
}
// ---- AUDIO (on-demand source: system, then microphone) -------------
if let Some(r) = col.ended.clone() {
rep.skip("AUDIO", &format!("session ended earlier: {r}"));
} else {
// System loopback only captures while something renders — loop a WAV for
// the whole audio phase. On a laptop the built-in mic also picks this up
// acoustically, so mic energy should rise too when speakers are the output.
#[cfg(windows)]
let _snd = std::process::Command::new("powershell")
.args([
"-NoProfile", "-WindowStyle", "Hidden", "-Command",
"$p=New-Object Media.SoundPlayer 'C:\\Windows\\Media\\Ring01.wav'; $p.PlayLooping(); Start-Sleep 16",
])
.spawn();
// Let playback actually start before switching the source.
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(600), |_| false).await;
// --- SYSTEM ("what you hear" — WASAPI render loopback) ---
let i_sys = col.audio_data.len();
let n0 = col.audio_pkts;
session.audio_set(1);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(5), |c| c.audio_pkts >= n0 + 20).await;
let sys_pkts = col.audio_pkts - n0;
let sys_packets: Vec<Vec<u8>> = col.audio_data[i_sys..].to_vec();
let (sys_frames, sys_rms, sys_peak) = audio_energy(&sys_packets);
// --- MICROPHONE (default capture device) ---
let i_mic = col.audio_data.len();
let n1 = col.audio_pkts;
session.audio_set(2);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(5), |c| c.audio_pkts >= n1 + 20).await;
let mic_pkts = col.audio_pkts - n1;
let mic_packets: Vec<Vec<u8>> = col.audio_data[i_mic..].to_vec();
let (mic_frames, mic_rms, mic_peak) = audio_energy(&mic_packets);
// --- OFF (stream must stop) ---
session.audio_set(0);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(400), |_| false).await;
let n_off = col.audio_pkts;
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(1200), |_| false).await;
let after_off = col.audio_pkts - n_off;
let detail = format!(
"SYSTEM={sys_pkts}pkts/{sys_frames}dec rms={sys_rms:.0} peak={sys_peak} | \
MIC={mic_pkts}pkts/{mic_frames}dec rms={mic_rms:.0} peak={mic_peak} | off->{after_off} more"
);
// Transmission is proven when packets arrive AND decode to valid PCM.
// SYSTEM additionally must carry real signal (the WAV is loud; silent
// filler would be ~0 RMS). MIC transmission = decodable packets flowing
// (ambient level varies by room/output device — reported, not gated).
let sys_ok = sys_pkts > 0 && sys_frames > 0 && sys_rms > 150.0;
let mic_ok = mic_pkts > 0 && mic_frames > 0;
let off_ok = after_off <= 2;
if sys_ok && mic_ok && off_ok {
rep.pass("AUDIO", &detail);
} else {
let mut why = Vec::new();
if !sys_ok { why.push("SYSTEM not transmitting real audio"); }
if !mic_ok { why.push("MIC not transmitting"); }
if !off_ok { why.push("did not stop on 'off'"); }
rep.fail("AUDIO", &format!("{} [{detail}]", why.join("; ")));
}
}
// ---- CHAT -----------------------------------------------------------
if let Some(r) = col.ended.clone() {
rep.skip("CHAT", &format!("session ended earlier: {r}"));
} else {
let before = col.chats.len();
session.chat("e2e ping");
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(2), |c| {
c.chats.len() > before
})
.await;
if let Some(r) = col.ended.clone() {
rep.fail("CHAT", &format!("session ended after chat: {r}"));
} else if col.chats.len() > before {
rep.pass(
"CHAT",
&format!("echo received: {:?}", col.chats.last().map(|(_, t)| t.clone())),
);
} else {
rep.pass("CHAT", "sent 'e2e ping'; no echo to sender (expected), session alive");
}
}
// ---- QUALITY --------------------------------------------------------
if let Some(r) = col.ended.clone() {
rep.skip("QUALITY", &format!("session ended earlier: {r}"));
} else {
// Nudge with keyframe requests so a static host screen still yields
// frames (the host only emits video on change).
let b1 = col.frames;
session.set_quality(30, 12000, 1.0); // "High" preset
for _ in 0..5 {
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(400), |_| false).await;
}
let d_high = col.frames - b1;
let b2 = col.frames;
session.set_quality(30, 6000, 1.0); // "Balanced" preset
for _ in 0..5 {
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(400), |_| false).await;
}
let d_bal = col.frames - b2;
if let Some(r) = col.ended.clone() {
rep.fail("QUALITY", &format!("session ended during set_quality: {r}"));
} else if d_high > 0 && d_bal > 0 {
rep.pass(
"QUALITY",
&format!("high:+{d_high} frames, balanced:+{d_bal} frames (stream kept flowing)"),
);
} else {
rep.fail(
"QUALITY",
&format!("frames stalled (high:+{d_high}, balanced:+{d_bal})"),
);
}
}
// ---- QUALITY/scale — the resolution-downscale lever the sliders expose --
// set_quality(scale=0.5) must HALVE the host's encoded frame. We read the
// dims the HOST reports via session.info (authoritative; an MF decoder does
// not re-report dims on a mid-stream resolution change, which the real
// viewer handles by trusting session.info).
if let Some(r) = col.ended.clone() {
rep.skip("QUALITY/scale", &format!("session ended earlier: {r}"));
} else {
let latest_wh = |c: &Collected| -> (u64, u64) {
c.infos
.iter()
.rev()
.find_map(|v| {
let w = v.get("w").and_then(|x| x.as_u64())?;
let h = v.get("h").and_then(|x| x.as_u64())?;
if w > 0 && h > 0 { Some((w, h)) } else { None }
})
.unwrap_or((0, 0))
};
session.set_quality(30, 8000, 1.0);
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(2), |_| false).await;
let (mut w1, mut h1) = latest_wh(&col);
if w1 == 0 {
if let Some((_, _, w, h)) = &col.connected {
w1 = *w as u64;
h1 = *h as u64;
}
}
let target = (w1 as f32 * 0.8) as u64;
session.set_quality(30, 8000, 0.5);
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(6), |c| {
let (w, _) = latest_wh(c);
w > 0 && w < target
})
.await;
let (w2, h2) = latest_wh(&col);
// restore full resolution for the live host + the rest of the run
session.set_quality(30, 12000, 1.0);
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(2), |_| false).await;
if w1 > 0 && w2 > 0 && w2 < target {
rep.pass("QUALITY/scale", &format!(
"host re-encoded at scale 0.5: {w1}x{h1} -> {w2}x{h2} (restored to full) — downscale lever works"));
} else {
rep.fail("QUALITY/scale", &format!(
"host did not downscale at scale 0.5: {w1}x{h1} -> {w2}x{h2}"));
}
}
// ---- MONITOR --------------------------------------------------------
if let Some(r) = col.ended.clone() {
rep.skip("MONITOR", &format!("session ended earlier: {r}"));
} else {
let mons = session.monitors().to_vec();
let kb = col.keyframes;
session.monitor_switch(0);
session.request_keyframe();
let got_kf =
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(3), |c| c.keyframes > kb)
.await;
let fb = col.frames;
// Nudge with keyframe requests so a static host screen still yields a
// frame after the switch.
for _ in 0..6 {
if col.frames > fb {
break;
}
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(350), |c| c.frames > fb).await;
}
let resumed = col.frames > fb;
if let Some(r) = col.ended.clone() {
rep.fail("MONITOR", &format!("session ended during monitor switch: {r}"));
} else if resumed {
rep.pass(
"MONITOR",
&format!(
"{} monitor(s) advertised; switched to 0, keyframe={}, frames resumed (+{})",
mons.len(),
got_kf,
col.frames - fb
),
);
} else {
rep.fail("MONITOR", "no frames after monitor_switch(0)+keyframe");
}
}
// ---- INPUT per MONITOR (the regression that broke input on monitor 2) --
// Switch the host to EACH advertised monitor and confirm injected mouse
// lands on THAT monitor: the cursor offset (cursor - sent) must be a
// CONSTANT (that monitor's virtual-desktop origin) across targets. Distinct
// target x's mean a stuck/mis-mapped cursor can't fake a cluster.
#[cfg(windows)]
if let Some(r) = col.ended.clone() {
rep.skip("INPUT/per-monitor", &format!("session ended earlier: {r}"));
} else {
let mons = session.monitors().to_vec();
if mons.len() < 2 {
rep.skip("INPUT/per-monitor", &format!("only {} monitor(s)", mons.len()));
} else {
session.set_quality(30, 8000, 1.0); // 1:1 -> sent coords are physical pixels
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(700), |_| false).await;
let mut all_ok = true;
let mut details: Vec<String> = Vec::new();
for m in &mons {
session.monitor_switch(m.i);
session.request_keyframe();
let fb = col.frames;
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(2), |c| c.frames > fb).await;
let (fw, fh) = (m.w as i32, m.h as i32);
let targets = [
(fw * 30 / 100, fh * 40 / 100),
(fw * 55 / 100, fh * 35 / 100),
(fw * 45 / 100, fh * 65 / 100),
(fw * 70 / 100, fh * 55 / 100),
];
let mut offs: Vec<(i32, i32)> = Vec::new();
for t in targets.iter() {
let c = move_and_settle(&session, &mut rx, &mut col, &mut dec, *t).await;
offs.push((c.0 - t.0, c.1 - t.1));
}
let tol = (fw / 30).max(16);
let (mut best, mut best_x, mut best_y) = (0usize, 0i32, 0i32);
for cand in offs.iter() {
let n = offs
.iter()
.filter(|o| (o.0 - cand.0).abs() <= tol && (o.1 - cand.1).abs() <= tol)
.count();
if n > best {
best = n;
best_x = cand.0;
best_y = cand.1;
}
}
let ok = best >= 2; // >=2 distinct-x targets at one offset = proof
all_ok &= ok;
details.push(format!(
"mon#{} origin~({},{}) {}/{}{}",
m.i, best_x, best_y, best, targets.len(), if ok { "" } else { " FAIL" }
));
}
session.monitor_switch(0); // restore
if all_ok {
rep.pass("INPUT/per-monitor", &format!("injected mouse lands on EVERY monitor: {}", details.join("; ")));
} else {
rep.fail("INPUT/per-monitor", &format!("input mis-maps on some monitor: {}", details.join("; ")));
}
}
}
// ---- SYS/processes --------------------------------------------------
if let Some(r) = col.ended.clone() {
rep.skip("SYS/processes", &format!("session ended earlier: {r}"));
} else {
let n0 = count_kind(&col, "processes");
session.list_processes();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| {
count_kind(c, "processes") > n0 && last_items_len(c, "processes").unwrap_or(0) > 0
})
.await;
match last_items_len(&col, "processes") {
Some(n) if n > 0 => rep.pass("SYS/processes", &format!("sys.res kind=processes, {n} items")),
Some(_) => rep.fail("SYS/processes", "processes reply had 0 items"),
None => rep.fail("SYS/processes", "no sys.res kind=processes within 4s"),
}
}
// ---- SYS/run (execute CMD/PowerShell on the host — admin feature) ---
if let Some(r) = col.ended.clone() {
rep.skip("SYS/run", &format!("session ended earlier: {r}"));
} else {
let n0 = count_kind(&col, "run");
session.run_command("echo NETDESK-EXEC-OK && whoami");
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(6), |c| {
count_kind(c, "run") > n0
})
.await;
let msg = last_str(&col, "run", "msg").unwrap_or_default();
let elevated = last_bool(&col, "run", "elevated");
if msg.contains("NETDESK-EXEC-OK") {
rep.pass(
"SYS/run",
&format!(
"host executed the command and returned output ({} bytes); elevated={:?}",
msg.len(),
elevated
),
);
} else if count_kind(&col, "run") > n0 {
rep.fail(
"SYS/run",
&format!("run reply arrived but marker missing; msg='{}'", msg.chars().take(80).collect::<String>()),
);
} else {
rep.fail("SYS/run", "no sys.res kind=run within 6s");
}
}
// ---- SYS/refresh_screen (forces a keyframe) -------------------------
if let Some(r) = col.ended.clone() {
rep.skip("SYS/refresh_screen", &format!("session ended earlier: {r}"));
} else {
let kb = col.keyframes;
session.refresh_screen();
let got =
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| c.keyframes > kb)
.await;
if got {
rep.pass("SYS/refresh_screen", &format!("keyframe received (+{})", col.keyframes - kb));
} else {
rep.fail("SYS/refresh_screen", "no keyframe within 4s");
}
}
// ---- SYS/block_input (toggle on then back off) ----------------------
if let Some(r) = col.ended.clone() {
rep.skip("SYS/block_input", &format!("session ended earlier: {r}"));
} else {
let n0 = count_kind(&col, "block_input");
session.block_input();
let got1 =
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| {
count_kind(c, "block_input") > n0
})
.await;
let on1 = last_on(&col, "block_input");
let n1 = count_kind(&col, "block_input");
session.block_input();
let got2 =
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| {
count_kind(c, "block_input") > n1
})
.await;
let on2 = last_on(&col, "block_input");
// SAFETY: never leave the host's input blocked.
if last_on(&col, "block_input") == Some(true) {
let n2 = count_kind(&col, "block_input");
session.block_input();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(3), |c| {
count_kind(c, "block_input") > n2
})
.await;
}
if got1 && got2 && on1 == Some(true) && on2 == Some(false) {
rep.pass("SYS/block_input", "on=true then on=false (input restored)");
} else {
rep.fail(
"SYS/block_input",
&format!("expected replies on=true then on=false, saw got1={got1} on1={on1:?} got2={got2} on2={on2:?}"),
);
}
}
// ---- SYS/hide_wallpaper (toggle then restore) -----------------------
if let Some(r) = col.ended.clone() {
rep.skip("SYS/hide_wallpaper", &format!("session ended earlier: {r}"));
} else {
let n0 = count_kind(&col, "wallpaper");
session.hide_wallpaper();
let got1 =
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| {
count_kind(c, "wallpaper") > n0
})
.await;
let on1 = last_on(&col, "wallpaper");
let n1 = count_kind(&col, "wallpaper");
session.hide_wallpaper();
let got2 =
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(4), |c| {
count_kind(c, "wallpaper") > n1
})
.await;
let on2 = last_on(&col, "wallpaper");
// SAFETY: never leave the wallpaper hidden.
if last_on(&col, "wallpaper") == Some(true) {
let n2 = count_kind(&col, "wallpaper");
session.hide_wallpaper();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(3), |c| {
count_kind(c, "wallpaper") > n2
})
.await;
}
if got1 && got2 && on2 == Some(false) {
rep.pass(
"SYS/hide_wallpaper",
&format!("sys.res kind=wallpaper (on1={on1:?} then on2=false, restored)"),
);
} else {
rep.fail(
"SYS/hide_wallpaper",
&format!("expected two wallpaper replies ending on=false, saw got1={got1} on1={on1:?} got2={got2} on2={on2:?}"),
);
}
}
// ---- FILES/list_roots ----------------------------------------------
let mut first_root: Option<String> = None;
if let Some(r) = col.ended.clone() {
rep.skip("FILES/list_roots", &format!("session ended earlier: {r}"));
} else {
let before = col.file_lists.len();
session.list_roots();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(5), |c| {
c.file_lists.len() > before
})
.await;
match col.file_lists.get(before).cloned() {
Some((_path, entries)) if !entries.is_empty() => {
first_root = entries.first().map(|e| e.name.clone());
let names: Vec<String> = entries.iter().map(|e| e.name.clone()).collect();
rep.pass(
"FILES/list_roots",
&format!("{} root(s): {}", entries.len(), names.join(", ")),
);
}
Some(_) => rep.fail("FILES/list_roots", "roots listing was empty"),
None => rep.fail("FILES/list_roots", "no FileList within 5s"),
}
}
// ---- FILES/list_dir (+ optional download) ---------------------------
if let Some(r) = col.ended.clone() {
rep.skip("FILES/list_dir", &format!("session ended earlier: {r}"));
rep.skip("FILES/download", "depends on list_dir");
} else if let Some(root) = first_root.clone() {
let before = col.file_lists.len();
session.list_dir(&root);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(5), |c| {
c.file_lists.len() > before
})
.await;
match col.file_lists.get(before).cloned() {
Some((_path, entries)) if !entries.is_empty() => {
rep.pass(
"FILES/list_dir",
&format!("{} entries under {}", entries.len(), root),
);
// optional: download the smallest regular file <= 256 KiB.
let small = entries
.iter()
.filter(|e| !e.is_dir && e.size > 0 && e.size <= 256 * 1024)
.min_by_key(|e| e.size)
.cloned();
if let Some(f) = small {
let full = join_path(&root, &f.name);
let ndl = col.downloads.len();
let nerr = col.file_errs.len();
session.download(&full);
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(8), |c| {
c.downloads.len() > ndl || c.file_errs.len() > nerr
})
.await;
if let Some((_fid, name, size, path)) = col.downloads.get(ndl).cloned() {
let len = std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
let _ = std::fs::remove_file(&path);
if len == size && size == f.size {
rep.pass(
"FILES/download",
&format!("{name}: {len} bytes on disk == host size {size} == listing size"),
);
} else {
rep.fail(
"FILES/download",
&format!(
"{name}: got {len} bytes, host size {size}, listing size {}",
f.size
),
);
}
} else if let Some(e) = col.file_errs.get(nerr) {
rep.fail("FILES/download", &format!("file.err: {e}"));
} else {
rep.fail("FILES/download", "no FileDownloadDone within 8s");
}
} else {
rep.skip(
"FILES/download",
&format!("no regular file <=256KiB in {root} to safely test"),
);
}
}
Some(_) => {
rep.fail("FILES/list_dir", &format!("{root} listed but 0 entries"));
rep.skip("FILES/download", "list_dir returned no entries");
}
None => {
rep.fail("FILES/list_dir", "no FileList within 5s");
rep.skip("FILES/download", "list_dir did not respond");
}
}
} else {
rep.skip("FILES/list_dir", "no root available from list_roots");
rep.skip("FILES/download", "no root available from list_roots");
}
// ---- FILES/upload (verified on disk — the live host writes here) -----
// This is the same path the viewer's "Upload…" button and drag & drop use.
if let Some(r) = col.ended.clone() {
rep.skip("FILES/upload", &format!("session ended earlier: {r}"));
} else {
let dir = std::env::temp_dir();
let fname = format!("netdesk-e2e-up-{}.txt", now_unix());
let content = format!("netdesk upload e2e {}", now_unix()).into_bytes();
let dest = dir.join(&fname);
let _ = std::fs::remove_file(&dest);
let dir_s = dir.to_string_lossy().to_string();
session.upload(&dir_s, &fname, &content);
let mut ok = false;
for _ in 0..25 {
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(120), |_| false).await;
if std::fs::read(&dest).map(|g| g == content).unwrap_or(false) {
ok = true;
break;
}
}
let _ = std::fs::remove_file(&dest);
if ok {
rep.pass("FILES/upload", &format!(
"uploaded {} bytes as {fname}; file on the host matches byte-for-byte (removed after)",
content.len()));
} else {
rep.fail("FILES/upload", "uploaded file never appeared on the host with matching content");
}
}
// ---- FILES/download-stream — multi-chunk, disk-streamed, byte-verified --
// Round-trip a 1 MiB file (4 x 256 KiB chunks) to exercise the sequential
// seek/write disk-streaming path (no RAM buffering) and prove integrity.
if let Some(r) = col.ended.clone() {
rep.skip("FILES/download-stream", &format!("session ended earlier: {r}"));
} else {
let dir = std::env::temp_dir();
let fname = format!("netdesk-e2e-stream-{}.bin", now_unix());
let hostfile = dir.join(&fname); // host writes here (host == this machine)
let _ = std::fs::remove_file(&hostfile);
let data: Vec<u8> = (0..1024u32 * 1024)
.map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
.collect();
session.upload(&dir.to_string_lossy(), &fname, &data);
let mut uploaded = false;
for _ in 0..60 {
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(120), |_| false).await;
if std::fs::metadata(&hostfile).map(|m| m.len() == data.len() as u64).unwrap_or(false) {
uploaded = true;
break;
}
}
// Settle so the host flushes/closes its write handle before we read the
// file back — otherwise the download can race the still-open writer and
// return same-length-but-different bytes.
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(800), |_| false).await;
if !uploaded {
rep.fail("FILES/download-stream", "upload of the 1 MiB test file never completed");
} else {
let ndl = col.downloads.len();
let nerr = col.file_errs.len();
session.download(&hostfile.to_string_lossy());
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(20), |c| {
c.downloads.len() > ndl || c.file_errs.len() > nerr
})
.await;
if let Some((_fid, _name, size, path)) = col.downloads.get(ndl).cloned() {
let got = std::fs::read(&path).unwrap_or_default();
let _ = std::fs::remove_file(&path);
if got.len() == data.len() && got == data {
rep.pass("FILES/download-stream", &format!(
"1 MiB round-trip ({} chunks) streamed to disk, byte-for-byte match (host size {size})",
data.len() / (256 * 1024)));
} else {
rep.fail("FILES/download-stream", &format!(
"mismatch: {} bytes vs {} (host size {size})", got.len(), data.len()));
}
} else if let Some(e) = col.file_errs.get(nerr) {
rep.fail("FILES/download-stream", &format!("file.err: {e}"));
} else {
rep.fail("FILES/download-stream", "no FileDownloadDone within 20s");
}
}
let _ = std::fs::remove_file(&hostfile);
}
// ---- remote cursor overlay (position + bitmap stream) ----------------
if col.cursors > 0 {
rep.pass(
"CURSOR",
&format!(
"{} remote-cursor positions, shape bitmap {} bytes",
col.cursors, col.cursor_shape_bytes
),
);
} else {
rep.fail("CURSOR", "no remote-cursor events received");
}
// ---- RECORDING — mux the live H.264 into a real .mp4, read it back -----
// The viewer's Rec button tees the same access units into this recorder.
if let Some(r) = col.ended.clone() {
rep.skip("RECORDING", &format!("session ended earlier: {r}"));
} else {
let (w, h) = col
.connected
.as_ref()
.map(|c| (c.2 as u16, c.3 as u16))
.unwrap_or((1920, 1080));
let path = std::env::temp_dir().join(format!("netdesk-e2e-rec-{}.mp4", now_unix()));
col.recorder = Some(netdesk_core::recorder::Mp4Recorder::new(&path, w, h));
session.set_quality(30, 8000, 1.0);
// Force several frames even if the host screen is static.
for _ in 0..8 {
session.request_keyframe();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_millis(400), |_| false).await;
}
let rec = col.recorder.take().unwrap();
let fed = rec.samples();
match rec.finish() {
Ok((p, _)) => {
let sz = std::fs::metadata(&p).map(|m| m.len()).unwrap_or(0);
match netdesk_core::recorder::probe_mp4(&p) {
Ok((vt, samples, pw, ph)) if vt >= 1 && samples > 0 => rep.pass(
"RECORDING",
&format!(
"valid .mp4 ({sz} bytes): {vt} video track, {samples} samples, {pw}x{ph} — H.264 remux, no re-encode"
),
),
Ok((vt, samples, _, _)) => rep.fail(
"RECORDING",
&format!("mp4 parsed but empty: {vt} tracks / {samples} samples (fed {fed})"),
),
Err(e) => rep.fail("RECORDING", &format!("mp4 unreadable: {e} (fed {fed})")),
}
let _ = std::fs::remove_file(&p);
}
Err(e) => rep.fail("RECORDING", &format!("finalize failed: {e}")),
}
}
// ---- close cleanly --------------------------------------------------
session.close();
pump_until(&mut rx, &mut col, &mut dec, Duration::from_secs(2), |c| c.ended.is_some()).await;
println!(
" session closed (ended reason: {})",
col.ended.as_deref().unwrap_or("<no Ended event>")
);
println!(
" tallies: frames_rx={} decoded={} keyframes={} audio_pkts={} sys_replies={} file_lists={}",
col.frames, col.decoded, col.keyframes, col.audio_pkts, col.sys.len(), col.file_lists.len()
);
summary(&rep);
}
fn host_ctl_checks(rep: &mut Report) {
// Point config resolution at the live engine's portable config (fresh
// heartbeat) unless the operator already set NETDESK_CONFIG.
if std::env::var_os("NETDESK_CONFIG").is_none() {
std::env::set_var("NETDESK_CONFIG", PORTABLE_CONFIG);
}
match host_ctl::read_config() {
Ok(cfg) => {
if cfg.id == host_id() {
rep.pass(
"HOSTCTL/read_config",
&format!("id={} name={} server={}", cfg.id, cfg.name, cfg.server),
);
} else {
rep.fail(
"HOSTCTL/read_config",
&format!("id={} (expected {})", cfg.id, host_id()),
);
}
let es = host_ctl::engine_status(&cfg);
if es.online {
rep.pass(
"HOSTCTL/engine_status",
&format!("online=true state={} in_session={}", es.state, es.in_session),
);
} else {
rep.fail(
"HOSTCTL/engine_status",
&format!("online=false state={}", es.state),
);
}
match host_ctl::ping_relay(&cfg) {
Ok(ms) => rep.pass("HOSTCTL/ping_relay", &format!("{ms} ms to {}/health", cfg.api_base())),
Err(e) => rep.fail("HOSTCTL/ping_relay", &format!("{e}")),
}
match host_ctl::list_devices(&cfg) {
Ok(devs) => {
let names: Vec<String> = devs
.iter()
.map(|d| format!("{}({},{})", d.name, d.id, if d.online { "online" } else { "offline" }))
.collect();
let has_devpc = devs.iter().any(|d| d.name.to_uppercase().contains("DEVPC"));
if !devs.is_empty() && has_devpc {
rep.pass(
"HOSTCTL/list_devices",
&format!("{} device(s): {}", devs.len(), names.join(", ")),
);
} else if !devs.is_empty() {
rep.fail(
"HOSTCTL/list_devices",
&format!("{} device(s) but none named DEVPC: {}", devs.len(), names.join(", ")),
);
} else {
rep.fail("HOSTCTL/list_devices", "0 devices returned");
}
}
Err(e) => rep.fail("HOSTCTL/list_devices", &format!("{e}")),
}
}
Err(e) => {
rep.fail("HOSTCTL/read_config", &format!("{e}"));
rep.fail("HOSTCTL/engine_status", "no config");
rep.fail("HOSTCTL/ping_relay", "no config");
rep.fail("HOSTCTL/list_devices", "no config");
}
}
}
fn summary(rep: &Report) {
println!("================================================================");
println!("WINDOW_END {}", now_unix());
println!(
"SUMMARY: {} / {} passed ({} skipped)",
rep.pass,
rep.pass + rep.fail,
rep.skip
);
}