順番

shiguredo · Nov 10, 2024 · 2efaba8 · 2efaba8
1 parent 8e4b777
commit 2efaba8
Showing 1 changed file with 78 additions and 38 deletions.
diff --git a/tests/test_sora_disconnect.py b/tests/test_sora_disconnect.py
@@ -10,43 +10,6 @@
 from sora_sdk import SoraSignalingErrorCode
 
 
-def test_websocket_signaling_only_lifetime_expired(setup):
-    signaling_urls = setup.get("signaling_urls")
-    channel_id_prefix = setup.get("channel_id_prefix")
-    secret = setup.get("secret")
-
-    channel_id = f"{channel_id_prefix}_{__name__}_{sys._getframe().f_code.co_name}_{uuid.uuid4()}"
-
-    access_token = jwt.encode(
-        {
-            "channel_id": channel_id,
-            "audio": False,
-            "video": True,
-            "connection_lifetime": 3,
-            # 現在時刻 + 300 秒 (5分)
-            "exp": int(time.time()) + 300,
-        },
-        secret,
-        algorithm="HS256",
-    )
-
-    with SoraClient(
-        signaling_urls,
-        SoraRole.RECVONLY,
-        channel_id,
-        audio=True,
-        video=True,
-        metadata={"access_token": access_token},
-        data_channel_signaling=False,
-        ignore_disconnect_websocket=False,
-    ) as conn:
-        time.sleep(5)
-
-        assert conn.ws_close is True
-        assert conn.ws_close_code == 1000
-        assert conn.ws_close_reason == "LIFETIME-EXPIRED"
-
-
 @pytest.mark.skipif(sys.platform != "linux", reason="linux でのみ実行する")
 def test_websocket_signaling_only_disconnect_api(setup):
     signaling_urls = setup.get("signaling_urls")
@@ -85,6 +48,43 @@ def test_websocket_signaling_only_disconnect_api(setup):
         # assert "DISCONNECTED-API" in conn.disconnect_reason
 
 
+def test_websocket_signaling_only_lifetime_expired(setup):
+    signaling_urls = setup.get("signaling_urls")
+    channel_id_prefix = setup.get("channel_id_prefix")
+    secret = setup.get("secret")
+
+    channel_id = f"{channel_id_prefix}_{__name__}_{sys._getframe().f_code.co_name}_{uuid.uuid4()}"
+
+    access_token = jwt.encode(
+        {
+            "channel_id": channel_id,
+            "audio": False,
+            "video": True,
+            "connection_lifetime": 3,
+            # 現在時刻 + 300 秒 (5分)
+            "exp": int(time.time()) + 300,
+        },
+        secret,
+        algorithm="HS256",
+    )
+
+    with SoraClient(
+        signaling_urls,
+        SoraRole.RECVONLY,
+        channel_id,
+        audio=True,
+        video=True,
+        metadata={"access_token": access_token},
+        data_channel_signaling=False,
+        ignore_disconnect_websocket=False,
+    ) as conn:
+        time.sleep(5)
+
+        assert conn.ws_close is True
+        assert conn.ws_close_code == 1000
+        assert conn.ws_close_reason == "LIFETIME-EXPIRED"
+
+
 @pytest.mark.skipif(sys.platform != "linux", reason="linux でのみ実行する")
 def test_websocket_datachannel_signaling_disconnect_api(setup):
     signaling_urls = setup.get("signaling_urls")
@@ -203,4 +203,44 @@ def test_datachannel_only_signaling_disconnect_api(setup):
         assert conn.disconnect_reason is not None
         assert "DISCONNECTED-API" in conn.disconnect_reason
 
-    # TODO: LIFETIME-EXPIRED のテスト
+
+def test_datachannel_only_signaling_lifetime_expired(setup):
+    signaling_urls = setup.get("signaling_urls")
+    channel_id_prefix = setup.get("channel_id_prefix")
+    secret = setup.get("secret")
+
+    channel_id = f"{channel_id_prefix}_{__name__}_{sys._getframe().f_code.co_name}_{uuid.uuid4()}"
+
+    access_token = jwt.encode(
+        {
+            "channel_id": channel_id,
+            "audio": True,
+            "video": True,
+            "connection_lifetime": 3,
+            # 現在時刻 + 300 秒 (5分)
+            "exp": int(time.time()) + 300,
+        },
+        secret,
+        algorithm="HS256",
+    )
+
+    with SoraClient(
+        signaling_urls,
+        SoraRole.RECVONLY,
+        channel_id,
+        audio=True,
+        video=True,
+        metadata={"access_token": access_token},
+        data_channel_signaling=True,
+        ignore_disconnect_websocket=True,
+    ) as conn:
+        time.sleep(5)
+
+        assert conn.close_message is not None
+        assert conn.close_message["type"] == "close"
+        assert conn.close_message["code"] == 1000
+        assert conn.close_message["reason"] == "LIFETIME-EXPIRED"
+
+        assert conn.disconnect_code == SoraSignalingErrorCode.CLOSE_SUCCEEDED
+        assert conn.disconnect_reason is not None
+        assert "LIFETIME-EXPIRED" in conn.disconnect_reason