lesavka/testing/tests/video_downstream_feed_contract.rs

//! Downstream eye-video feed contracts.
//!
//! Scope: exercise the server eye stream with a deterministic GStreamer test
//! source and lock down the native source-mode policy used by the launcher and
//! diagnostics.
//! Targets: `server/src/video.rs`, `server/src/video_support.rs`,
//! `common/src/eye_source.rs`.
//! Why: the live capture cards are hardware-dependent, but this catches the
//! packet, timing, keyframe, and mode-selection regressions that break the
//! client previews before hardware enters the loop.

use futures_util::StreamExt;
use lesavka_common::eye_source::{
    EyeSourceMode, default_eye_source_mode, display_size_for_source_mode,
    eye_source_mode_for_request, native_eye_source_modes,
};
use lesavka_server::{video, video_support::contains_idr};
use serial_test::serial;
use std::time::Duration;

#[test]
fn native_downstream_eye_modes_stay_widescreen_and_square_pixel() {
    assert_eq!(
        native_eye_source_modes(),
        &[
            EyeSourceMode {
                width: 1920,
                height: 1080,
                fps: 60,
            },
            EyeSourceMode {
                width: 1280,
                height: 720,
                fps: 60,
            },
        ]
    );
    assert_eq!(
        default_eye_source_mode(),
        EyeSourceMode {
            width: 1920,
            height: 1080,
            fps: 60,
        }
    );
    for mode in native_eye_source_modes() {
        assert_eq!(
            display_size_for_source_mode(*mode),
            (mode.width, mode.height)
        );
    }
}

#[test]
fn downstream_eye_request_chooses_best_native_capture_mode() {
    assert_eq!(
        eye_source_mode_for_request(0, 0),
        EyeSourceMode {
            width: 1920,
            height: 1080,
            fps: 60,
        }
    );
    assert_eq!(
        eye_source_mode_for_request(1920, 1080),
        EyeSourceMode {
            width: 1920,
            height: 1080,
            fps: 60,
        }
    );
    assert_eq!(
        eye_source_mode_for_request(1600, 900),
        EyeSourceMode {
            width: 1280,
            height: 720,
            fps: 60,
        }
    );
    assert_eq!(
        eye_source_mode_for_request(960, 540),
        EyeSourceMode {
            width: 1280,
            height: 720,
            fps: 60,
        }
    );
}

#[test]
#[serial]
fn testsrc_downstream_feed_emits_h264_for_both_eyes() {
    with_fast_video_env(|| {
        let runtime = tokio::runtime::Runtime::new().expect("runtime");
        runtime.block_on(async {
            assert_testsrc_downstream_eye(0, 1920, 1080, 60).await;
            assert_testsrc_downstream_eye(1, 1280, 720, 60).await;
        });
    });
}

async fn assert_testsrc_downstream_eye(eye_id: u32, width: u32, height: u32, fps: u32) {
    let setup = tokio::time::timeout(
        Duration::from_secs(2),
        video::eye_ball_with_request("testsrc", eye_id, 1_200, width, height, fps),
    )
    .await;
    let mut stream = match setup {
        Ok(Ok(stream)) => stream,
        Ok(Err(err)) => panic!("testsrc eye-{eye_id} setup failed: {err:#}"),
        Err(_) => panic!("testsrc eye-{eye_id} setup timed out"),
    };

    let mut last_pts = None;
    let mut saw_annex_b = false;
    let mut saw_idr = false;
    for packet_index in 0..8 {
        let packet = tokio::time::timeout(Duration::from_secs(2), stream.next())
            .await
            .unwrap_or_else(|_| panic!("eye-{eye_id} packet {packet_index} timed out"))
            .expect("stream should stay open")
            .unwrap_or_else(|err| panic!("eye-{eye_id} packet {packet_index} failed: {err}"));

        assert_eq!(packet.id, eye_id, "logical eye id must be preserved");
        assert!(
            !packet.data.is_empty(),
            "eye-{eye_id} packet {packet_index} must carry H.264 bytes"
        );
        if let Some(previous) = last_pts {
            assert!(
                packet.pts >= previous,
                "eye-{eye_id} PTS must not move backward"
            );
        }
        last_pts = Some(packet.pts);
        saw_annex_b |= contains_annex_b_start_code(&packet.data);
        saw_idr |= contains_idr(&packet.data);
    }

    assert!(saw_annex_b, "eye-{eye_id} should emit Annex-B H.264");
    assert!(
        saw_idr,
        "eye-{eye_id} should emit an IDR keyframe for decoder recovery"
    );
    drop(stream);
}

fn with_fast_video_env(run: impl FnOnce()) {
    temp_env::with_vars(
        [
            ("LESAVKA_EYE_ADAPTIVE", Some("0")),
            ("LESAVKA_EYE_QUEUE_BUFFERS", Some("8")),
            ("LESAVKA_EYE_APPSINK_BUFFERS", Some("8")),
            ("LESAVKA_EYE_CHAN_CAPACITY", Some("32")),
            ("LESAVKA_EYE_TESTSRC_KBIT", Some("1200")),
            ("LESAVKA_EYE_KEYFRAME_INTERVAL", Some("1")),
        ],
        run,
    );
}

fn contains_annex_b_start_code(bytes: &[u8]) -> bool {
    bytes.windows(3).any(|window| window == [0, 0, 1])
        || bytes.windows(4).any(|window| window == [0, 0, 0, 1])
}