Files
livekit-cameras/tests/test_latency_flags.py
root 95bb7c50ee feat: point the fleet at wss LiveKit room uwh-telhai
Publish and subscribe over wss://livekit.uni-wh.de:7800 and refuse
cleartext ws://. Conference room is uwh-telhai. Includes the uncommitted
encoded H.264 publish path, Rally hairpin, and KMS wall overlay.
2026-10-11 00:48:50 +00:00

65 lines
2.1 KiB
Python

"""Env-flagged low-latency knobs (hop/queue, hold, enhance scope)."""
from latency import (
audio_queue_size_ms,
effective_enhance_mode,
video_hold_seconds,
video_stream_kwargs,
)
def test_audio_queue_defaults_to_legacy_floor():
assert audio_queue_size_ms(0.1, audio_queue_ms=None) == 150
assert audio_queue_size_ms(0.2, audio_queue_ms=None) == 250
def test_audio_queue_env_overrides_floor():
assert audio_queue_size_ms(0.1, audio_queue_ms=60) == 60
assert audio_queue_size_ms(0.05, audio_queue_ms=60) == 60
def test_video_hold_matches_hop_when_audio_present():
assert video_hold_seconds(0.1, has_audio=True, video_hold_s=None) == 0.1
assert video_hold_seconds(0.05, has_audio=True, video_hold_s=None) == 0.05
def test_video_hold_zero_without_audio():
assert video_hold_seconds(0.1, has_audio=False, video_hold_s=None) == 0.0
assert video_hold_seconds(0.1, has_audio=False, video_hold_s=0.2) == 0.0
def test_video_hold_env_overrides_including_zero():
assert video_hold_seconds(0.1, has_audio=True, video_hold_s=0.0) == 0.0
assert video_hold_seconds(0.1, has_audio=True, video_hold_s=0.05) == 0.05
def test_enhance_speaker_only_rally_disables_c920s():
assert effective_enhance_mode(
"auto", "cam-01", speaker_only=True, speaker_camera="rally",
) == "off"
assert effective_enhance_mode(
"auto", "rally", speaker_only=True, speaker_camera="rally",
) == "auto"
def test_enhance_speaker_only_inactive_leaves_mode():
assert effective_enhance_mode(
"auto", "cam-07", speaker_only=False, speaker_camera="cam-01",
) == "auto"
def test_enhance_speaker_only_ignored_when_speaker_is_active():
assert effective_enhance_mode(
"auto", "cam-07", speaker_only=True, speaker_camera="active",
) == "auto"
def test_video_stream_legacy_unbounded_native():
assert video_stream_kwargs(capacity=0, format_name="") == {}
def test_video_stream_latest_bgra():
assert video_stream_kwargs(capacity=1, format_name="bgra") == {
"capacity": 1,
"format": "bgra",
}