lesavka/server/src/upstream_media_runtime/tests/async_wait.rs

use super::{UpstreamMediaRuntime, play};
use std::sync::Arc;
use std::time::Duration;

#[tokio::test(flavor = "current_thread")]
async fn wait_for_audio_master_releases_video_once_audio_catches_up() {
    let runtime = Arc::new(UpstreamMediaRuntime::new());
    let _camera = runtime.activate_camera();
    let _microphone = runtime.activate_microphone();

    assert!(matches!(
        runtime.plan_video_pts(1_000_000, 16_666),
        super::UpstreamPlanDecision::AwaitingPair
    ));
    let _audio_first = play(runtime.plan_audio_pts(1_000_000));
    let video_first = play(runtime.plan_video_pts(1_000_000, 16_666));

    let waiter = tokio::spawn({
        let runtime = runtime.clone();
        async move {
            runtime
                .wait_for_audio_master(video_first.local_pts_us + 10_000, video_first.due_at)
                .await
        }
    });

    tokio::time::sleep(Duration::from_millis(5)).await;
    let _audio_next = play(runtime.plan_audio_pts(1_010_000));

    assert!(waiter.await.expect("audio master waiter should finish"));
}

#[tokio::test(flavor = "current_thread")]
async fn wait_for_audio_master_times_out_when_audio_never_catches_up() {
    let runtime = Arc::new(UpstreamMediaRuntime::new());
    let _camera = runtime.activate_camera();
    let _microphone = runtime.activate_microphone();

    assert!(matches!(
        runtime.plan_video_pts(1_000_000, 16_666),
        super::UpstreamPlanDecision::AwaitingPair
    ));
    let _audio_first = play(runtime.plan_audio_pts(1_000_000));
    let video_first = play(runtime.plan_video_pts(1_000_000, 16_666));

    let due_at = tokio::time::Instant::now() + Duration::from_millis(20);
    assert!(
        !runtime
            .wait_for_audio_master(video_first.local_pts_us + 100_000, due_at)
            .await
    );
}

#[tokio::test(flavor = "current_thread")]
async fn wait_for_audio_master_returns_true_when_no_microphone_stream_is_active() {
    let runtime = Arc::new(UpstreamMediaRuntime::new());
    let camera = runtime.activate_camera();
    let microphone = runtime.activate_microphone();
    runtime.close_microphone(microphone.generation);

    assert!(runtime.is_camera_active(camera.generation));
    assert!(
        runtime
            .wait_for_audio_master(
                123_456,
                tokio::time::Instant::now() + Duration::from_millis(10)
            )
            .await
    );
}

#[tokio::test(flavor = "current_thread")]
async fn new_microphone_owner_waits_for_the_previous_sink_to_release() {
    let runtime = Arc::new(UpstreamMediaRuntime::new());
    let first = runtime.activate_microphone();
    let first_permit = runtime
        .reserve_microphone_sink(first.generation)
        .await
        .expect("first owner should acquire the sink gate");
    let second = runtime.activate_microphone();

    let waiter = tokio::spawn({
        let runtime = runtime.clone();
        async move {
            runtime
                .reserve_microphone_sink(second.generation)
                .await
                .is_some()
        }
    });

    tokio::time::sleep(Duration::from_millis(25)).await;
    assert!(!waiter.is_finished());

    drop(first_permit);
    assert!(waiter.await.expect("waiter task should finish"));
}

#[tokio::test(flavor = "current_thread")]
async fn superseded_microphone_waiter_stands_down_before_opening_a_sink() {
    let runtime = Arc::new(UpstreamMediaRuntime::new());
    let first = runtime.activate_microphone();
    let first_permit = runtime
        .reserve_microphone_sink(first.generation)
        .await
        .expect("first owner should acquire the sink gate");
    let second = runtime.activate_microphone();

    let superseded_waiter = tokio::spawn({
        let runtime = runtime.clone();
        async move {
            runtime
                .reserve_microphone_sink(second.generation)
                .await
                .is_some()
        }
    });

    tokio::time::sleep(Duration::from_millis(25)).await;
    let _third = runtime.activate_microphone();
    drop(first_permit);

    assert!(
        !superseded_waiter
            .await
            .expect("superseded waiter task should finish"),
        "older waiter should stand down instead of opening a sink after supersession"
    );
}