signaling forwarding_filter(s) のテストを追加

shiguredo · Oct 28, 2024 · 1d2c461 · 1d2c461
1 parent 7843559
commit 1d2c461
Show file tree

Hide file tree

Showing 2 changed files with 102 additions and 0 deletions.
diff --git a/tests/client.py b/tests/client.py
@@ -47,6 +47,8 @@ def __init__(
         data_channel_signaling: Optional[bool] = None,
         ignore_disconnect_websocket: Optional[bool] = None,
         data_channels: Optional[list[dict[str, Any]]] = None,
+        forwarding_filter: Optional[dict[str, Any]] = None,
+        forwarding_filters: Optional[list[dict[str, Any]]] = None,
         openh264_path: Optional[str] = None,
         client_key: Optional[bytes] = None,
         client_cert: Optional[bytes] = None,
@@ -118,6 +120,8 @@ def __init__(
             data_channel_signaling=self._connect_data_channel_signaling,
             ignore_disconnect_websocket=self._connect_ignore_disconnect_websocket,
             data_channels=self._connect_data_channels,
+            forwarding_filter=forwarding_filter,
+            forwarding_filters=forwarding_filters,
             audio_source=self._audio_source,
             video_source=self._video_source,
             ca_cert=ca_cert,

diff --git a/tests/test_signaling_message.py b/tests/test_signaling_message.py
@@ -35,3 +35,101 @@ def test_signaling_message(setup):
     assert sendonly.connect_message["audio"] is True
     assert sendonly.connect_message["video"] is True
     assert sendonly.connect_message["metadata"] == metadata
+
+
+def test_signaling_message_type_connect_forwarding_filter(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()}"
+
+    forwarding_filter = {
+        "name": "test",
+        "priority": 128,
+        "action": "block",
+        "rules": [
+            [
+                {
+                    "field": "connection_id",
+                    "operator": "is_in",
+                    "values": ["S8YEN0TSE13JDC2991NG4XZ150"],
+                }
+            ],
+            [
+                {"field": "client_id", "operator": "is_in", "values": ["screen-share"]},
+                {"field": "kind", "operator": "is_in", "values": ["audio"]},
+            ],
+        ],
+    }
+
+    sendonly = SoraClient(
+        signaling_urls,
+        SoraRole.SENDONLY,
+        channel_id,
+        audio=True,
+        video=True,
+        metadata=metadata,
+        forwarding_filter=forwarding_filter,
+    )
+    sendonly.connect(fake_audio=True, fake_video=True)
+
+    time.sleep(5)
+
+    sendonly.disconnect()
+
+    assert sendonly.connect_message is not None
+    assert sendonly.connect_message["forwarding_filter"] == forwarding_filter
+
+
+def test_signaling_message_type_connect_forwarding_filters(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()}"
+
+    forwarding_filters = [
+        {
+            "name": "test",
+            "priority": 128,
+            "action": "block",
+            "rules": [
+                [
+                    {
+                        "field": "connection_id",
+                        "operator": "is_in",
+                        "values": ["S8YEN0TSE13JDC2991NG4XZ150"],
+                    }
+                ],
+                [
+                    {"field": "client_id", "operator": "is_in", "values": ["screen-share"]},
+                    {"field": "kind", "operator": "is_in", "values": ["audio"]},
+                ],
+            ],
+        },
+        {
+            "name": "test2",
+            "priority": 129,
+            "action": "block",
+            "rules": [[{"field": "kind", "operator": "is_in", "values": ["audio", "video"]}]],
+        },
+    ]
+
+    sendonly = SoraClient(
+        signaling_urls,
+        SoraRole.SENDONLY,
+        channel_id,
+        audio=True,
+        video=True,
+        metadata=metadata,
+        forwarding_filters=forwarding_filters,
+    )
+    sendonly.connect(fake_audio=True, fake_video=True)
+
+    time.sleep(5)
+
+    sendonly.disconnect()
+
+    assert sendonly.connect_message is not None
+    assert sendonly.connect_message["forwarding_filters"] == forwarding_filters