44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
"""Smoke test: one camera worker against the local LiveKit server, ~20 s."""
|
|
import json
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).parent
|
|
sys.path.insert(0, str(HERE))
|
|
|
|
from discovery import discover_cameras
|
|
from tokens import make_token
|
|
|
|
cam = discover_cameras()[0]
|
|
tok = make_token("ws://localhost:7880", "devkey", "devsecret", "smoke", "smoke-test")
|
|
|
|
p = subprocess.Popen(
|
|
[str(HERE / ".venv/bin/python"), str(HERE / "run_worker.py"),
|
|
"--camera-json", json.dumps({
|
|
"index": 0, "identity": "smoke-test",
|
|
"video_device": cam.video_device, "audio_card": cam.audio_card,
|
|
"usb_port": cam.usb_port}),
|
|
"--room", "smoke", "--url", "ws://localhost:7880", "--token", tok,
|
|
"--width", "640", "--height", "480", "--fps", "30",
|
|
"--audio-rate", "16000", "--enhance-mode", "off"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
|
|
)
|
|
t0 = time.time()
|
|
lines = []
|
|
while time.time() - t0 < 20:
|
|
line = p.stdout.readline()
|
|
if line:
|
|
lines.append(line.rstrip())
|
|
print(line.rstrip(), flush=True)
|
|
p.send_signal(signal.SIGTERM)
|
|
try:
|
|
rc = p.wait(timeout=15)
|
|
except subprocess.TimeoutExpired:
|
|
p.kill()
|
|
rc = "killed"
|
|
print(f"--- smoke test done: rc={rc}")
|
|
print("published markers:", [l for l in lines if "published" in l or "status" in l or "UP" in l])
|