root
95bb7c50ee
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.
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
from config import Config
|
|
from discovery import Camera
|
|
from run import worker_args_for
|
|
import json
|
|
import jwt
|
|
|
|
|
|
def _cfg(**kw) -> Config:
|
|
return Config(url="ws://127.0.0.1:7880", api_key="k", api_secret="s", **kw)
|
|
|
|
|
|
def _cam(identity: str = "cam-07") -> Camera:
|
|
return Camera(index=0, identity=identity, video_device="/dev/cam07", audio_card=2)
|
|
|
|
|
|
def test_worker_args_legacy_omits_optional_latency_flags():
|
|
args = worker_args_for(_cfg(), _cam())
|
|
assert "--audio-queue-ms" not in args
|
|
assert "--video-hold-s" not in args
|
|
assert "--split-executor" not in args
|
|
assert args[args.index("--enhance-mode") + 1] == "auto"
|
|
|
|
|
|
def test_worker_args_latency_flags_and_non_speaker_enhance_off():
|
|
cfg = _cfg(
|
|
audio_queue_ms=60,
|
|
video_hold_s=0.0,
|
|
worker_split_executor=True,
|
|
enhance_speaker_only=True,
|
|
display_speaker_camera="cam-01",
|
|
enhance_mode="auto",
|
|
)
|
|
args = worker_args_for(cfg, _cam("cam-07"))
|
|
assert args[args.index("--audio-queue-ms") + 1] == "60"
|
|
assert args[args.index("--video-hold-s") + 1] == "0.0"
|
|
assert "--split-executor" in args
|
|
assert args[args.index("--enhance-mode") + 1] == "off"
|
|
assert args[args.index("--beamform-mode") + 1] == "off"
|
|
|
|
|
|
def test_worker_args_speaker_keeps_enhance_mode():
|
|
cfg = _cfg(
|
|
enhance_mode="force",
|
|
enhance_speaker_only=True,
|
|
display_speaker_camera="cam-07",
|
|
)
|
|
args = worker_args_for(cfg, _cam("cam-07"))
|
|
assert args[args.index("--enhance-mode") + 1] == "force"
|
|
assert args[args.index("--beamform-mode") + 1] == "auto"
|
|
|
|
|
|
def test_worker_token_carries_uwh_tag():
|
|
args = worker_args_for(_cfg(), _cam("cam-07"))
|
|
token = args[args.index("--token") + 1]
|
|
claims = jwt.decode(token, "s", algorithms=["HS256"])
|
|
assert claims["attributes"]["tag"] == "uwh"
|
|
assert claims["attributes"]["tags"] == "uwh"
|
|
|
|
|
|
def test_worker_args_passes_enhance_socket():
|
|
args = worker_args_for(_cfg(enhance_socket="/tmp/e.sock"), _cam())
|
|
assert args[args.index("--enhance-socket") + 1] == "/tmp/e.sock"
|
|
|
|
|
|
def test_worker_args_omits_enhance_socket_when_empty():
|
|
args = worker_args_for(_cfg(enhance_socket=""), _cam())
|
|
assert "--enhance-socket" not in args
|
|
|
|
|
|
def test_worker_args_passes_video_shm():
|
|
args = worker_args_for(_cfg(), _cam("cam-07"))
|
|
assert args[args.index("--video-shm") + 1] == "/run/livekit-cameras/raw/cam-07.i420"
|
|
|
|
|
|
def test_worker_args_omits_video_shm_when_capture_daemon_off():
|
|
args = worker_args_for(_cfg(capture_daemon=False), _cam())
|
|
assert "--video-shm" not in args
|
|
|
|
|
|
def test_publisher_args_single_process_and_speaker_only():
|
|
from run import publisher_args_for
|
|
cfg = _cfg(enhance_speaker_only=True, display_speaker_camera="cam-01")
|
|
args = publisher_args_for(cfg, [_cam("cam-07"), _cam("cam-01")])
|
|
assert "--enhance-speaker-only" in args
|
|
assert "--audio-gate" not in args
|
|
payload = json.loads(args[args.index("--cameras-json") + 1])
|
|
assert {p["identity"] for p in payload} == {"cam-07", "cam-01"}
|
|
assert all("token" in p for p in payload)
|
|
assert args[args.index("--video-shm-dir") + 1] == "/run/livekit-cameras/raw"
|
|
assert args[args.index("--audio-shm-dir") + 1] == "/run/livekit-cameras/pcm"
|
|
|
|
|
|
def test_publisher_args_rally_keeps_1080p_dims():
|
|
from run import publisher_args_for
|
|
rally = Camera(index=20, identity="rally", video_device="/dev/rally",
|
|
audio_card=0, width=1920, height=1080)
|
|
args = publisher_args_for(_cfg(), [_cam("cam-01"), rally])
|
|
payload = json.loads(args[args.index("--cameras-json") + 1])
|
|
by_id = {p["identity"]: p for p in payload}
|
|
assert by_id["rally"]["width"] == 1920
|
|
assert by_id["rally"]["height"] == 1080
|
|
assert by_id["cam-01"]["width"] is None
|
|
|