test_abort.py を追加

tests/api.py

@@ -1,5 +1,14 @@
 import httpx
 
+
+def disconnect_channel_api(url: str, channel_id: str) -> httpx.Response:
+    headers = {
+        "Content-Type": "application/json",
+        "x-sora-target": "Sora_20151104.DisconnectChannel",
+    }
+    return httpx.post(url, headers=headers, json={"channel_id": channel_id}, follow_redirects=True)
+
+
 def disconnect_connection_api(url: str, channel_id: str, connection_id: str) -> httpx.Response:
     headers = {
         "Content-Type": "application/json",

tests/test_abort.py

@@ -0,0 +1,37 @@
+import sys
+import uuid
+
+from api import disconnect_channel_api
+from client import SoraClient, SoraRole
+
+
+def test_abort(setup):
+    """
+    abort させるためのテスト
+    """
+    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()}"
+
+    conn1 = SoraClient(
+        signaling_urls,
+        SoraRole.SENDONLY,
+        channel_id,
+        audio=True,
+        video=True,
+        metadata=metadata,
+    ).connect(fake_audio=True, fake_video=True)
+
+    conn2 = SoraClient(
+        signaling_urls,
+        SoraRole.RECVONLY,
+        channel_id,
+        audio=True,
+        video=True,
+        metadata=metadata,
+    ).connect()
+
+    response = disconnect_channel_api(signaling_urls, channel_id)
+    assert response.status_code == 200, [response.text]