テスト書き直し

tests/test_messaging.py

@@ -10,8 +10,6 @@
 
 
 class Messaging:
-    connection_created = False
-
     def __init__(self, signaling_urls, channel_id, label, direction, metadata):
         self.sora = Sora()
         self.connection = self.sora.create_connection(
@@ -25,7 +23,8 @@ def __init__(self, signaling_urls, channel_id, label, direction, metadata):
             data_channel_signaling=True,
         )
 
-        self.disconnected = False
+        self.connected = False
+        self.closed = False
         self.label = label
         self.is_data_channel_ready = False
         self.connection.on_set_offer = self.on_set_offer
@@ -47,11 +46,11 @@ def on_notify(self, raw_message):
             and message["connection_id"] == self.connection_id
         ):
             print(f"Sora に接続しました: connection_id={self.connection_id}")
-            self.connection_created = True
+            self.connected = True
 
     def on_disconnect(self, error_code, message):
         print(f"Sora から切断しました: error_code='{error_code}' message='{message}'")
-        self.disconnected = True
+        self.closed = True
 
     def on_message(self, label, data):
         print(f"メッセージを受信しました: label={label}, data={data}")
@@ -65,7 +64,7 @@ def connect(self):
 
     def send(self, data):
         # on_data_channel() が呼ばれるまではデータチャネルの準備ができていないので待機
-        while not self.is_data_channel_ready and not self.disconnected:
+        while not self.is_data_channel_ready and not self.closed:
             time.sleep(0.01)
 
         self.connection.send_data_channel(self.label, data)
@@ -81,7 +80,7 @@ def sendonly(signaling_urls, channel_id, label, metadata):
 
     time.sleep(3)
 
-    assert msg_sendonly.connection_created is True
+    assert msg_sendonly.connected is True
 
     msg_sendonly.connection.send_data_channel(label, b"Hello, world!")
 
@@ -96,7 +95,7 @@ def recvonly(signaling_urls, channel_id, label, metadata):
 
     time.sleep(3)
 
-    assert msg_recvonly.connection_created is True
+    assert msg_recvonly.connected is True
 
     time.sleep(3)
 
@@ -112,16 +111,16 @@ def test_messaging_direction_recvonly():
     msg_recvonly = Messaging(signaling_urls, channel_id, label, "recvonly", metadata)
     msg_sendonly = Messaging(signaling_urls, channel_id, label, "sendonly", metadata)
 
-    assert msg_recvonly.connection_created is False
-    assert msg_sendonly.connection_created is False
+    assert msg_recvonly.connected is False
+    assert msg_sendonly.connected is False
 
     msg_recvonly.connect()
     msg_sendonly.connect()
 
     time.sleep(3)
 
-    assert msg_recvonly.connection_created is True
-    assert msg_sendonly.connection_created is True
+    assert msg_recvonly.connected is True
+    assert msg_sendonly.connected is True
 
     msg_sendonly.send(b"Hello, world!")
 

tests/test_recvonly.py

@@ -1,29 +1,80 @@
+import json
 import os
 import time
 
+from dotenv import load_dotenv
 from sora_sdk import Sora
 
+load_dotenv()
 
-def on_disconnect(error_code, message: str):
-    print(f"on_disconnect: error_code: {error_code}, message: {message}")
 
+class Recvonly:
+    def __init__(self, signaling_urls, channel_id, metadata):
+        self.sora = Sora()
+        self.connection = self.sora.create_connection(
+            signaling_urls=signaling_urls,
+            role="sendrecv",
+            channel_id=channel_id,
+            metadata=metadata,
+        )
 
-def on_notify(raw_message: str):
-    print(f'on_notify: raw_message: {raw_message}')
+        self.connected = False
+        self.closed = False
+        self.is_data_channel_ready = False
 
+        self.connection.on_set_offer = self.on_set_offer
+        self.connection.on_notify = self.on_notify
+        self.connection.on_disconnect = self.on_disconnect
 
-def test_sendonly():
-    sora = Sora()
+        self.connection_id = None
 
-    conn = sora.create_connection(
-        signaling_urls=[os.environ.get("TEST_SIGNALING_URL")],
-        role="recvonly",
-        channel_id=os.environ.get("TEST_CHANNEL_ID_PREFIX") + "sora-python-sdk-test",
-        metadata={"access_token": os.environ.get("TEST_SECRET_KEY")},
-    )
+    def on_set_offer(self, raw_offer):
+        offer = json.loads(raw_offer)
+        if offer["type"] == "offer":
+            self.connection_id = offer["connection_id"]
+
+    def on_notify(self, raw_message):
+        message = json.loads(raw_message)
+        if (
+            message["type"] == "notify"
+            and message["event_type"] == "connection.created"
+            and message["connection_id"] == self.connection_id
+        ):
+            print(f"Sora に接続しました: connection_id={self.connection_id}")
+            self.connected = True
+
+    def on_disconnect(self, error_code, message):
+        print(f"Sora から切断しました: error_code='{error_code}' message='{message}'")
+        self.closed = True
+
+    def connect(self):
+        self.connection.connect()
+
+    def disconnect(self):
+        self.connection.disconnect()
+
+
+def test_recvonly():
+    signaling_urls = [os.environ.get("TEST_SIGNALING_URL")]
+    channel_id = os.environ.get("TEST_CHANNEL_ID_PREFIX") + "sora-python-sdk-test"
+    metadata = {"access_token": os.environ.get("TEST_SECRET_KEY")}
+
+    recvonly = Recvonly(signaling_urls, channel_id, metadata)
+
+    assert recvonly.connected is False
+
+    recvonly.connect()
 
-    conn.on_notify = on_notify
-    conn.on_disconnect = on_disconnect
-    conn.connect()
     time.sleep(3)
-    conn.disconnect()
+
+    assert recvonly.connected is True
+
+    time.sleep(3)
+
+    assert recvonly.closed is False
+
+    recvonly.disconnect()
+
+    time.sleep(3)
+
+    assert recvonly.closed is True