lesavka/client/src/live_capture_clock/tests.rs

use super::{
    DurationPacedSourcePtsRebaser, SourcePtsRebaser, capture_pts_us, packet_age,
    upstream_source_lag_cap, upstream_source_lead_cap, upstream_timing_trace_enabled,
};
use serial_test::serial;
use std::time::Duration;

#[test]
#[serial]
fn capture_pts_us_monotonically_advances() {
    let first = capture_pts_us();
    std::thread::sleep(Duration::from_millis(2));
    let second = capture_pts_us();
    assert!(second >= first);
}

#[test]
#[serial]
fn packet_age_is_small_for_recent_packets() {
    let pts = capture_pts_us();
    std::thread::sleep(Duration::from_millis(2));
    let age = packet_age(pts);
    assert!(age >= Duration::from_millis(1));
    assert!(age < Duration::from_secs(1));
}

#[test]
#[serial]
fn source_pts_rebaser_preserves_source_delta_on_shared_capture_clock() {
    let rebased = SourcePtsRebaser::default();
    let first = rebased.rebase_or_now(Some(1_000_000), 1);
    let second = rebased.rebase_or_now(Some(1_033_333), 1);

    assert!(first.used_source_pts);
    assert_eq!(
        second.packet_pts_us.saturating_sub(first.packet_pts_us),
        33_333
    );
    assert_eq!(first.source_base_us, Some(1_000_000));
    assert_eq!(second.source_base_us, Some(1_000_000));
    assert_eq!(first.capture_base_us, second.capture_base_us);
}

#[test]
#[serial]
fn source_pts_rebaser_stays_monotonic_when_source_pts_repeat() {
    let rebased = SourcePtsRebaser::default();
    let first = rebased.rebase_or_now(Some(50_000), 1);
    let second = rebased.rebase_or_now(Some(50_000), 1);

    assert_eq!(second.packet_pts_us, first.packet_pts_us + 1);
}

#[test]
#[serial]
fn source_pts_rebaser_falls_back_to_capture_clock_without_source_pts() {
    let rebased = SourcePtsRebaser::default();
    let first = rebased.rebase_or_now(None, 1);
    std::thread::sleep(Duration::from_millis(2));
    let second = rebased.rebase_or_now(None, 1);

    assert!(!first.used_source_pts);
    assert!(!second.used_source_pts);
    assert!(second.packet_pts_us > first.packet_pts_us);
    assert!(!first.lag_clamped);
    assert!(!second.lag_clamped);
    assert!(!first.lead_clamped);
    assert!(!second.lead_clamped);
}

#[test]
#[serial]
fn source_pts_rebaser_clamps_source_lag_when_it_falls_too_far_behind_now() {
    let rebased = SourcePtsRebaser::default();
    let _first = rebased.rebase_with_lag_cap(Some(1_000_000), 1, None);
    std::thread::sleep(Duration::from_millis(8));
    let second = rebased.rebase_with_lag_cap(Some(1_000_001), 1, Some(Duration::from_millis(2)));

    assert!(second.used_source_pts);
    assert!(second.lag_clamped);
    assert!(second.capture_now_us >= second.packet_pts_us);
    assert!(second.capture_now_us - second.packet_pts_us <= 2_500);
}

#[test]
#[serial]
fn source_pts_rebaser_clamps_source_lead_when_it_runs_too_far_ahead() {
    temp_env::with_var("LESAVKA_UPSTREAM_SOURCE_LEAD_CAP_MS", Some("5"), || {
        let rebased = SourcePtsRebaser::default();
        let _first =
            rebased.rebase_with_lag_cap(Some(1_000_000), 1, Some(Duration::from_millis(250)));
        let second =
            rebased.rebase_with_lag_cap(Some(2_000_000), 1, Some(Duration::from_millis(250)));

        assert!(second.used_source_pts);
        assert!(second.lead_clamped);
        assert!(second.packet_pts_us >= second.capture_now_us);
        assert!(second.packet_pts_us <= second.capture_now_us + 5_500);
    });
}

#[test]
#[serial]
fn source_pts_rebasers_anchor_each_stream_to_its_own_first_packet_time() {
    let microphone = SourcePtsRebaser::default();
    let camera = SourcePtsRebaser::default();

    let first_microphone = microphone.rebase_or_now(Some(80_000), 1);
    std::thread::sleep(Duration::from_millis(5));
    let first_camera = camera.rebase_or_now(Some(435_000), 1);

    assert_ne!(
        first_microphone.capture_base_us, first_camera.capture_base_us,
        "independent camera/mic pipelines must not be forced onto the same first-packet timestamp"
    );
    assert!(
        first_camera.packet_pts_us > first_microphone.packet_pts_us,
        "a later-starting camera pipeline should keep that real wall-clock delay"
    );
    assert_eq!(first_microphone.source_base_us, Some(80_000));
    assert_eq!(first_camera.source_base_us, Some(435_000));
}

#[test]
#[serial]
fn upstream_timing_trace_flag_defaults_off_and_accepts_true_values() {
    temp_env::with_var_unset("LESAVKA_UPSTREAM_TIMING_TRACE", || {
        assert!(!upstream_timing_trace_enabled());
    });

    temp_env::with_var("LESAVKA_UPSTREAM_TIMING_TRACE", Some("1"), || {
        assert!(upstream_timing_trace_enabled());
    });

    temp_env::with_var("LESAVKA_UPSTREAM_TIMING_TRACE", Some("false"), || {
        assert!(!upstream_timing_trace_enabled());
    });
}

#[test]
#[serial]
fn upstream_source_lag_cap_defaults_and_accepts_override() {
    temp_env::with_var_unset("LESAVKA_UPSTREAM_SOURCE_LAG_CAP_MS", || {
        assert_eq!(upstream_source_lag_cap(), Duration::from_millis(250));
    });

    temp_env::with_var("LESAVKA_UPSTREAM_SOURCE_LAG_CAP_MS", Some("90"), || {
        assert_eq!(upstream_source_lag_cap(), Duration::from_millis(90));
    });
}

#[test]
#[serial]
fn upstream_source_lead_cap_defaults_and_accepts_override() {
    temp_env::with_var_unset("LESAVKA_UPSTREAM_SOURCE_LEAD_CAP_MS", || {
        assert_eq!(upstream_source_lead_cap(), Duration::from_millis(80));
    });

    temp_env::with_var("LESAVKA_UPSTREAM_SOURCE_LEAD_CAP_MS", Some("35"), || {
        assert_eq!(upstream_source_lead_cap(), Duration::from_millis(35));
    });
}

#[test]
#[serial]
fn duration_paced_rebaser_advances_by_packet_duration_when_source_pts_stretch() {
    let rebased = DurationPacedSourcePtsRebaser::default();
    let first = rebased.rebase_with_packet_duration(Some(0), 21_333, Duration::from_millis(250));
    let second =
        rebased.rebase_with_packet_duration(Some(52_666), 21_333, Duration::from_millis(250));

    assert_eq!(
        second.packet_pts_us.saturating_sub(first.packet_pts_us),
        21_333
    );
}

#[test]
#[serial]
fn duration_paced_rebaser_clamps_when_duration_pacing_falls_stale() {
    let rebased = DurationPacedSourcePtsRebaser::default();
    let _first = rebased.rebase_with_packet_duration(Some(0), 10_000, Duration::from_millis(2));
    std::thread::sleep(Duration::from_millis(8));
    let second =
        rebased.rebase_with_packet_duration(Some(10_000), 10_000, Duration::from_millis(2));

    assert!(
        second.packet_pts_us.saturating_add(2_500) >= second.capture_now_us,
        "duration-paced packet pts should never trail live capture by more than the lag cap"
    );
}

#[test]
#[serial]
fn duration_paced_rebaser_clamps_when_duration_pacing_runs_future() {
    temp_env::with_var("LESAVKA_UPSTREAM_SOURCE_LEAD_CAP_MS", Some("15"), || {
        let rebased = DurationPacedSourcePtsRebaser::default();
        let mut last =
            rebased.rebase_with_packet_duration(Some(0), 50_000, Duration::from_millis(250));
        for packet_index in 1..12 {
            last = rebased.rebase_with_packet_duration(
                Some(packet_index * 50_000),
                50_000,
                Duration::from_millis(250),
            );
        }

        assert!(last.lead_clamped);
        assert!(last.packet_pts_us <= last.capture_now_us + 16_000);
    });
}