simulcast の送信テストを追加

tests/client.py

@@ -37,13 +37,18 @@ def __init__(
  use_hwa: bool = False,
  audio_channels: int = 1,
  audio_sample_rate: int = 16000,
+ video_width: int = 640,
+ video_height: int = 480,
  ):
  self._signaling_urls: list[str] = signaling_urls
  self._channel_id: str = channel_id
 
  self._audio_channels: int = audio_channels
  self._audio_sample_rate: int = audio_sample_rate
 
+ self._video_width: int = video_width
+ self._video_height: int = video_height
+
  self._sora: Sora = Sora(openh264=openh264_path, use_hardware_encoder=use_hwa)
 
  self._fake_audio_thread: Optional[threading.Thread] = None
@@ -162,7 +167,9 @@ def _fake_audio_loop(self):
  def _fake_video_loop(self):
  while not self._closed.is_set():
  time.sleep(1.0 / 30)
- self._video_source.on_captured(numpy.zeros((480, 640, 3), dtype=numpy.uint8))
+ self._video_source.on_captured(
+ numpy.zeros((self._video_height, self._video_width, 3), dtype=numpy.uint8)
+ )
 
  def _on_signaling_message(
  self,

tests/test_simulcast.py

@@ -0,0 +1,54 @@
+import sys
+import time
+import uuid
+
+from client import Sendonly
+
+
+def test_simulcast_vp8(setup):
+ signaling_urls = setup.get("signaling_urls")
+ channel_id_prefix = setup.get("channel_id_prefix")
+ metadata = setup.get("metadata")
+
+ channel_id = f"{channel_id_prefix}_{__name__}_{sys._getframe().f_code.co_name}_{uuid.uuid4()}"
+
+ sendonly = Sendonly(
+ signaling_urls,
+ channel_id,
+ simulcast=True,
+ audio=False,
+ video=True,
+ video_codec_type="VP8",
+ video_bit_rate=3000,
+ metadata=metadata,
+ video_width=1280,
+ video_height=720,
+ )
+ sendonly.connect(fake_video=True)
+
+ time.sleep(5)
+
+ sendonly_stats = sendonly.get_stats()
+
+ sendonly.disconnect()
+
+ # codec が無かったら StopIteration 例外が上がる
+ sendonly_codec_stats = next(s for s in sendonly_stats if s.get("type") == "codec")
+ assert sendonly_codec_stats["mimeType"] == "video/VP8"
+
+ # 複数のoutbound-rtp統計情報を取得
+ outbound_rtp_stats = [
+ s for s in sendonly_stats if s.get("type") == "outbound-rtp" and s.get("kind") == "video"
+ ]
+ assert len(outbound_rtp_stats) == 3
+
+ # rid でソート
+ sorted_stats = sorted(outbound_rtp_stats, key=lambda x: x.get("rid", ""))
+
+ for i, rtp_stat in enumerate(sorted_stats):
+ assert rtp_stat["rid"] == f"r{i}"
+ assert (
+ rtp_stat["encoderImplementation"] == "SimulcastEncoderAdapter (libvpx, libvpx, libvpx)"
+ )
+ assert rtp_stat["bytesSent"] > 0
+ assert rtp_stat["packetsSent"] > 0