0080 E2E の AV1 ペアテストを Intel VPL デコーダーが使えない環境で xfail にする

CHANGES.md

@@ -17,6 +17,16 @@
   - Sora Labo の JWT `turn_tls_only` で TURN-TLS を強制し、metrics の `relayProtocol` が `tls` であることを確認する
   - @Hexa
 
+### misc
+
+- [UPDATE] Intel VPL の E2E テストで、AV1 デコーダーが検出されない環境では AV1 のペアテストを xfail にする
+  - `--video-codec-engines` の出力で AV1 デコーダーに Intel VPL が含まれるかどうかを判定する
+  - @voluntas
+- [UPDATE] テストコードの型エラーを修正する
+  - ty の型検査が通るように TypedDict を追加する
+  - 型チェッカーの ignore 指定を ty の形式に修正する
+  - @voluntas
+
 ## 2026.1.0
 
 **リリース日**: 2026-09-16

issues/0080-change-e2e-av1-xfail.md → issues/closed/0080-change-e2e-av1-xfail.md

@@ -1,7 +1,7 @@
 # E2E の AV1 ペアテストを Intel VPL デコーダーが使えない環境で xfail にする
 
 - Created: 2026-09-30
-- Completed: {YYYY-MM-DD}
+- Completed: 2026-09-30
 - Branch: feature/change-e2e-av1-xfail
 - Polished: 2026-09-30
 
@@ -38,4 +38,10 @@ Intel VPL の AV1 デコーダーが検出されない環境で Intel VPL E2E 
 
 ## 解決方法
 
-{YYYY-MM-DD} に追記する
+2026-09-30 に解決した。
+
+- `test/momo.py` の `Momo` に `get_video_codec_engines()` を追加し、`--video-codec-engines` の出力からコーデックごとのエンコーダー / デコーダーの表示名を取得できるようにした
+- `test/test_sora_mode_intel_vpl.py` の `test_sora_sendonly_recvonly_pair` の AV1 のときだけ、AV1 デコーダーに `Intel VPL` が含まれない場合に実行時 `pytest.xfail(...)` を呼ぶようにした。一覧が空の場合は解析失敗として `RuntimeError` にし、無言の xfail にはしない
+- prek の ty チェックを通すため、`test/momo.py` に `VideoEncoderParams` / `VideoDecoderParams` を追加してコーデックエンジン指定の dict に型注釈を付け、各テストに適用した。`test_ayame_mode.py` の型チェッカーの ignore 指定を ty の形式に修正した
+- `CHANGES.md` の `## develop` に misc のエントリを追記した
+- 検証: `cd test && uv run ty check .` と `prek run` が通ることを確認した。xfail の実機動作は Intel VPL の self-hosted runner での CI 実行で確認する

test/momo.py

@@ -8,7 +8,7 @@
 from enum import StrEnum
 from pathlib import Path
 from types import TracebackType
-from typing import Any, Literal, Self
+from typing import Any, Literal, Self, TypedDict
 
 import httpx
 
@@ -21,6 +21,34 @@ class MomoMode(StrEnum):
     SORA = "sora"
 
 
+class VideoEncoderParams(TypedDict, total=False):
+    """Momo のキーワード引数へ `**` 展開して渡すビデオエンコーダーの指定
+
+    テスト側で dict を組み立てて `Momo(..., **encoder_params)` のように展開する。
+    キーと値の型をここで定義しておくことで、`**dict` 展開の型検査を通す。
+    """
+
+    vp8_encoder: Literal["default", "software"]
+    vp9_encoder: Literal["default", "vpl", "software"]
+    av1_encoder: Literal["default", "vpl", "nvidia", "software"]
+    h264_encoder: Literal["default", "vpl", "nvidia", "videotoolbox", "software"]
+    h265_encoder: Literal["default", "vpl", "nvidia", "videotoolbox"]
+
+
+class VideoDecoderParams(TypedDict, total=False):
+    """Momo のキーワード引数へ `**` 展開して渡すビデオデコーダーの指定
+
+    テスト側で dict を組み立てて `Momo(..., **decoder_params)` のように展開する。
+    キーと値の型をここで定義しておくことで、`**dict` 展開の型検査を通す。
+    """
+
+    vp8_decoder: Literal["default", "software"]
+    vp9_decoder: Literal["default", "vpl", "nvidia", "software"]
+    av1_decoder: Literal["default", "vpl", "nvidia", "software"]
+    h264_decoder: Literal["default", "vpl", "nvidia", "videotoolbox"]
+    h265_decoder: Literal["default", "vpl", "nvidia", "videotoolbox"]
+
+
 class Momo:
     """Momo プロセスを管理するクラス"""
 
@@ -76,7 +104,8 @@ def __init__(
         metrics_allow_external_ip: bool = False,
         client_cert: str | None = None,  # PEM ファイルパス
         client_key: str | None = None,  # PEM ファイルパス
-        ca_cert: str | None = None,  # PEM ファイルパス（指定時はその PEM のみを trust anchor にする）
+        ca_cert: str
+        | None = None,  # PEM ファイルパス（指定時はその PEM のみを trust anchor にする）
         proxy_url: str | None = None,
         proxy_username: str | None = None,
         proxy_password: str | None = None,
@@ -313,6 +342,54 @@ def _get_momo_executable_path(self) -> str:
 
         return str(momo_path)
 
+    def get_video_codec_engines(self) -> dict[str, dict[str, list[str]]]:
+        """momo の --video-codec-engines 出力を解析して利用できるエンジン一覧を返す
+
+        戻り値は {コーデック名: {"Encoder": [表示名, ...], "Decoder": [表示名, ...]}} の形式。
+        オプション値 (例: vpl) は将来変更される可能性があるため、環境の判定には
+        表示名 (例: Intel VPL) を使う。
+        """
+        result = subprocess.run(
+            [self.executable_path, "--video-codec-engines"],
+            capture_output=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=30,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise RuntimeError(
+                f"momo --video-codec-engines failed with code {result.returncode}: {result.stderr}"
+            )
+
+        engines: dict[str, dict[str, list[str]]] = {}
+        codec_name: str | None = None
+        section_name: str | None = None
+        for line in result.stdout.splitlines():
+            if not line:
+                continue
+            if not line.startswith(" "):
+                # コーデック名の行 (例: "AV1:")
+                codec_name = line.removesuffix(":")
+                engines[codec_name] = {"Encoder": [], "Decoder": []}
+                section_name = None
+                continue
+            stripped_line = line.strip()
+            if stripped_line in ("Encoder:", "Decoder:"):
+                # エンコーダー / デコーダーのセクション行
+                section_name = stripped_line.removesuffix(":")
+                continue
+            if codec_name is None or section_name is None:
+                continue
+            if stripped_line == "*UNAVAILABLE*":
+                # 利用できるエンジンが無いコーデックはエンジン行が出力されない
+                continue
+            if stripped_line.startswith("- "):
+                # エンジン行 (例: "- Intel VPL [vpl] (default)")
+                engine_name = stripped_line.removeprefix("- ").split(" [", 1)[0]
+                engines[codec_name][section_name].append(engine_name)
+        return engines
+
     def __enter__(self) -> Self:
         """コンテキストマネージャーの開始"""
         try:

test/test_ayame_mode.py

@@ -236,7 +236,7 @@ def test_ayame_mode_peer_connection(port_allocator):
             resolution="QVGA",
             initial_wait=10,
         ) as m2:
-            # 両方のピアの接続が確立されるまで弅機
+            # 両方のピアの接続が確立されるまで待機
             assert m1.wait_for_connection(timeout=10), (
                 "M1 failed to establish connection within timeout"
             )
@@ -384,27 +384,26 @@ def test_ayame_mode_with_invalid_codec(port_allocator):
     """存在しないコーデックを指定した場合にエラーで終了することを確認"""
     room_id = str(uuid.uuid4())
 
-    # 存在しないビデオコーデックを指定
-    # 型チェッカーの警告を抑制するため type: ignore コメントを使用
+    # 存在しないビデオコーデックを指定する
     with pytest.raises(RuntimeError, match="momo process exited unexpectedly"):
         with Momo(
             mode=MomoMode.AYAME,
             ayame_signaling_url=AYAME_SIGNALING_URL,
             room_id=room_id,
             metrics_port=next(port_allocator),
             fake_capture_device=True,
-            ayame_video_codec_type="INVALID_CODEC",  # type: ignore[arg-type] 存在しないコーデック
+            ayame_video_codec_type="INVALID_CODEC",  # ty: ignore[invalid-argument-type]
         ):
             pass  # ここには到達しないはず
 
-    # 存在しないオーディオコーデックを指定
+    # 存在しないオーディオコーデックを指定する
     with pytest.raises(RuntimeError, match="momo process exited unexpectedly"):
         with Momo(
             mode=MomoMode.AYAME,
             ayame_signaling_url=AYAME_SIGNALING_URL,
             room_id=room_id,
             metrics_port=next(port_allocator),
             fake_capture_device=True,
-            ayame_audio_codec_type="INVALID_AUDIO",  # type: ignore[arg-type] 存在しないコーデック
+            ayame_audio_codec_type="INVALID_AUDIO",  # ty: ignore[invalid-argument-type]
         ):
             pass  # ここには到達しないはず

test/test_sora_mode_apple_video_toolbox.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from momo import Momo, MomoMode
+from momo import Momo, MomoMode, VideoDecoderParams, VideoEncoderParams
 
 # Sora モードのテストは TEST_SORA_MODE_SIGNALING_URLS が設定されていない場合スキップ
 # Apple Video Toolbox 環境が有効でない場合もスキップ
@@ -26,7 +26,7 @@ def test_connection_stats(sora_settings, video_codec_type, free_port):
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "H264":
         encoder_params["h264_encoder"] = "videotoolbox"
     elif video_codec_type == "H265":
@@ -159,7 +159,7 @@ def test_connection_stats(sora_settings, video_codec_type, free_port):
 def test_simulcast(sora_settings, video_codec_type, expected_encoder_implementation, free_port):
     """Sora モードで simulcast 接続時の統計情報を確認（Apple Video Toolbox 使用）"""
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "H264":
         encoder_params["h264_encoder"] = "videotoolbox"
     elif video_codec_type == "H265":
@@ -419,14 +419,14 @@ def test_sora_sendonly_recvonly_pair(
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "H264":
         encoder_params["h264_encoder"] = "videotoolbox"
     elif video_codec_type == "H265":
         encoder_params["h265_encoder"] = "videotoolbox"
 
     # デコーダー設定を準備
-    decoder_params = {}
+    decoder_params: VideoDecoderParams = {}
     if video_codec_type == "H264":
         decoder_params["h264_decoder"] = "videotoolbox"
     elif video_codec_type == "H265":

test/test_sora_mode_intel_vpl.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from momo import Momo, MomoMode
+from momo import Momo, MomoMode, VideoDecoderParams, VideoEncoderParams
 
 # Sora モードのテストは TEST_SORA_MODE_SIGNALING_URLS が設定されていない場合スキップ
 # Intel VPL 環境が有効でない場合もスキップ
@@ -27,7 +27,7 @@ def test_connection_stats(sora_settings, video_codec_type, free_port):
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     match video_codec_type:
         case "VP9":
             encoder_params["vp9_encoder"] = "vpl"
@@ -195,7 +195,7 @@ def test_connection_stats(sora_settings, video_codec_type, free_port):
 def test_simulcast(sora_settings, video_codec_type, expected_encoder_implementation, free_port):
     """Sora モードで simulcast 接続時の統計情報を確認（Intel VPL 使用）"""
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     match video_codec_type:
         case "VP9":
             encoder_params["vp9_encoder"] = "vpl"
@@ -465,7 +465,7 @@ def test_sora_sendonly_recvonly_pair(
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     match video_codec_type:
         case "VP9":
             encoder_params["vp9_encoder"] = "vpl"
@@ -477,7 +477,7 @@ def test_sora_sendonly_recvonly_pair(
             encoder_params["h265_encoder"] = "vpl"
 
     # デコーダー設定を準備
-    decoder_params = {}
+    decoder_params: VideoDecoderParams = {}
     match video_codec_type:
         case "VP9":
             decoder_params["vp9_decoder"] = "vpl"
@@ -489,7 +489,7 @@ def test_sora_sendonly_recvonly_pair(
             decoder_params["h265_decoder"] = "vpl"
 
     # 送信専用クライアント
-    with Momo(
+    sender = Momo(
         mode=MomoMode.SORA,
         signaling_urls=sora_settings.signaling_urls,
         channel_id=sora_settings.channel_id,
@@ -502,7 +502,23 @@ def test_sora_sendonly_recvonly_pair(
         metadata=sora_settings.metadata,
         initial_wait=10,
         **encoder_params,
-    ) as sender:
+    )
+
+    # 一部の環境では momo が Intel VPL の AV1 デコーダーを検出できず、--av1-decoder に
+    # vpl を指定できない。検出できるようになるまでの一時的な措置として、デコーダーが
+    # 使えない環境では xfail にする。
+    if video_codec_type == "AV1":
+        av1_decoders = sender.get_video_codec_engines().get("AV1", {}).get("Decoder", [])
+        # 正常な環境では AV1 のソフトウェアデコーダーが必ず含まれるため、一覧が空の場合は
+        # --video-codec-engines の出力を解析できていない。無言の xfail にはしない。
+        if not av1_decoders:
+            raise RuntimeError(
+                "--video-codec-engines の出力から AV1 デコーダーの一覧を取得できない"
+            )
+        if "Intel VPL" not in av1_decoders:
+            pytest.xfail("この環境では momo が Intel VPL の AV1 デコーダーを検出できない")
+
+    with sender:
         # 受信専用クライアント
         with Momo(
             mode=MomoMode.SORA,

test/test_sora_mode_nvidia_video_codec.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from momo import Momo, MomoMode
+from momo import Momo, MomoMode, VideoDecoderParams, VideoEncoderParams
 
 # Sora モードのテストは TEST_SORA_MODE_SIGNALING_URLS が設定されていない場合スキップ
 # NVIDIA Video Codec 環境が有効でない場合もスキップ
@@ -26,7 +26,7 @@ def test_connection_stats(sora_settings, video_codec_type, free_port):
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "AV1":
         encoder_params["av1_encoder"] = "nvidia"
     elif video_codec_type == "H264":
@@ -158,7 +158,7 @@ def test_connection_stats(sora_settings, video_codec_type, free_port):
 def test_simulcast(sora_settings, video_codec_type, expected_encoder_implementation, free_port):
     """Sora モードで simulcast 接続時の統計情報を確認（NVIDIA Video Codec 使用）"""
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "AV1":
         encoder_params["av1_encoder"] = "nvidia"
     elif video_codec_type == "H264":
@@ -421,7 +421,7 @@ def test_sora_sendonly_recvonly_pair(
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "AV1":
         encoder_params["av1_encoder"] = "nvidia"
     elif video_codec_type == "H264":
@@ -430,7 +430,7 @@ def test_sora_sendonly_recvonly_pair(
         encoder_params["h265_encoder"] = "nvidia"
 
     # デコーダー設定を準備
-    decoder_params = {}
+    decoder_params: VideoDecoderParams = {}
     if video_codec_type == "AV1":
         decoder_params["av1_decoder"] = "nvidia"
     elif video_codec_type == "H264":

test/test_sora_mode_sendonly_recvonly.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from momo import Momo, MomoMode
+from momo import Momo, MomoMode, VideoDecoderParams, VideoEncoderParams
 
 # Sora モードのテストは TEST_SORA_MODE_SIGNALING_URLS が設定されていない場合スキップ
 pytestmark = pytest.mark.skipif(
@@ -34,7 +34,7 @@ def test_sendonly_recvonly_pair(
     expected_mime_type = f"video/{video_codec_type}"
 
     # エンコーダー設定を準備
-    encoder_params = {}
+    encoder_params: VideoEncoderParams = {}
     if video_codec_type == "VP8":
         encoder_params["vp8_encoder"] = "software"
     elif video_codec_type == "VP9":
@@ -43,7 +43,7 @@ def test_sendonly_recvonly_pair(
         encoder_params["av1_encoder"] = "software"
 
     # デコーダー設定を準備
-    decoder_params = {}
+    decoder_params: VideoDecoderParams = {}
     if video_codec_type == "VP8":
         decoder_params["vp8_decoder"] = "software"
     elif video_codec_type == "VP9":