chore: Add shared helper functions used across the livekit-rtc test suite.

Author: cloudwebrtc

livekit-rtc/tests/__init__.py livekit-rtc/tests/conftest.pylivekit-rtc/tests/__init__.py renamed to livekit-rtc/tests/conftest.py

@@ -11,3 +11,17 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
+"""Put this tests directory on sys.path so test modules can `from utils import ...`.
+
+Both `livekit-rtc/tests/` and the repo-root `tests/` are collected by pytest
+together; if either has an `__init__.py` they collide on the `tests` namespace
+package name. So we keep this directory non-package and use absolute imports.
+"""
+
+import sys
+from pathlib import Path
+
+_TESTS_DIR = Path(__file__).resolve().parent
+if str(_TESTS_DIR) not in sys.path:
+    sys.path.insert(0, str(_TESTS_DIR))

livekit-rtc/tests/test_audio.py

@@ -14,20 +14,23 @@
 
 """End-to-end audio publish/subscribe tests."""
 
+from __future__ import annotations
+
 import asyncio
 import ctypes
 import math
 import os
-import uuid
 import wave
 from pathlib import Path
 
 import numpy as np
 import pytest
 
-from livekit import api, rtc
+from livekit import rtc
 from livekit.rtc.audio_frame import AudioFrame
 
+from utils import create_token, skip_if_no_credentials, unique_room_name
+
 
 SAMPLE_RATE = 48000
 NUM_CHANNELS = 1
@@ -36,33 +39,6 @@
 AMPLITUDE = 0.5
 
 
-def skip_if_no_credentials():
-    required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
-    missing = [var for var in required_vars if not os.getenv(var)]
-    return pytest.mark.skipif(
-        bool(missing), reason=f"Missing environment variables: {', '.join(missing)}"
-    )
-
-
-def create_token(identity: str, room_name: str) -> str:
-    return (
-        api.AccessToken()
-        .with_identity(identity)
-        .with_name(identity)
-        .with_grants(
-            api.VideoGrants(
-                room_join=True,
-                room=room_name,
-            )
-        )
-        .to_jwt()
-    )
-
-
-def unique_room_name(base: str) -> str:
-    return f"{base}-{uuid.uuid4().hex[:8]}"
-
-
 def _generate_sine_wave(
     frequency: int,
     sample_rate: int,
@@ -132,7 +108,7 @@ def _band_energies(
 class TestAudioStreamPublishSubscribe:
     """End-to-end: publish a sine sweep into a room and verify spectrum on the subscriber."""
 
-    async def test_audio_stream_publish_subscribe(self):
+    async def test_audio_stream_publish_subscribe(self) -> None:
         """Publish 5 seconds of 100/300/500/700/1000 Hz tones and FFT-verify received audio."""
         url = os.environ["LIVEKIT_URL"]
         room_name = unique_room_name("test-audio-sweep")
@@ -151,7 +127,7 @@ def on_track_subscribed(
             track: rtc.Track,
             publication: rtc.RemoteTrackPublication,
             participant: rtc.RemoteParticipant,
-        ):
+        ) -> None:
             nonlocal subscribed_track
             if track.kind == rtc.TrackKind.KIND_AUDIO:
                 subscribed_track = track

livekit-rtc/tests/test_change_video_quality.py

@@ -42,16 +42,17 @@
 import os
 import sys
 import time
-import uuid
-from typing import Callable, Optional, Tuple
+from typing import Any, Callable, Optional, Tuple
 
 import numpy as np
 import pytest
 
-from livekit import api, rtc
+from livekit import rtc
 from livekit.rtc._proto.track_publication_pb2 import VideoQuality
 from livekit.rtc.room import EventTypes
 
+from utils import create_token, skip_if_no_credentials, unique_room_name
+
 
 WAIT_TIMEOUT = 30.0
 WAIT_INTERVAL = 0.1
@@ -73,33 +74,6 @@
 ]
 
 
-def skip_if_no_credentials():
-    required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
-    missing = [var for var in required_vars if not os.getenv(var)]
-    return pytest.mark.skipif(
-        bool(missing), reason=f"Missing environment variables: {', '.join(missing)}"
-    )
-
-
-def create_token(identity: str, room_name: str) -> str:
-    return (
-        api.AccessToken()
-        .with_identity(identity)
-        .with_name(identity)
-        .with_grants(
-            api.VideoGrants(
-                room_join=True,
-                room=room_name,
-            )
-        )
-        .to_jwt()
-    )
-
-
-def unique_room_name(base: str) -> str:
-    return f"{base}-{uuid.uuid4().hex[:8]}"
-
-
 async def _wait_until(
     predicate: Callable[[], bool],
     *,
@@ -149,7 +123,7 @@ def _expect_event(
     loop = asyncio.get_running_loop()
     fut: asyncio.Future = loop.create_future()
 
-    def _on_event(*args, **kwargs) -> None:
+    def _on_event(*args: Any, **kwargs: Any) -> None:
         if fut.done():
             return
         if predicate is None or predicate(*args, **kwargs):

livekit-rtc/tests/test_e2ee_per_participant.py

@@ -18,12 +18,17 @@
 
 import asyncio
 import os
-import uuid
-from typing import Any, Callable, TypeVar
 
 import pytest
 
-from livekit import api, rtc
+from livekit import rtc
+
+from utils import (
+    assert_eventually,
+    create_token,
+    skip_if_no_credentials,
+    unique_room_name,
+)
 
 
 # Per-participant keys (publisher identity → list of (key_bytes, key_index))
@@ -37,51 +42,6 @@
 WIDTH, HEIGHT = 320, 180
 FRAME_RATE = 15
 
-T = TypeVar("T")
-
-
-def skip_if_no_credentials() -> Any:
-    required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
-    missing = [var for var in required_vars if not os.getenv(var)]
-    return pytest.mark.skipif(
-        bool(missing), reason=f"Missing environment variables: {', '.join(missing)}"
-    )
-
-
-def create_token(identity: str, room_name: str) -> str:
-    return (
-        api.AccessToken()
-        .with_identity(identity)
-        .with_name(identity)
-        .with_grants(
-            api.VideoGrants(
-                room_join=True,
-                room=room_name,
-            )
-        )
-        .to_jwt()
-    )
-
-
-def unique_room_name(base: str) -> str:
-    return f"{base}-{uuid.uuid4().hex[:8]}"
-
-
-async def assert_eventually(
-    condition: Callable[[], T],
-    timeout: float = 15.0,
-    interval: float = 0.1,
-    message: str = "Condition not met within timeout",
-) -> T:
-    deadline = asyncio.get_event_loop().time() + timeout
-    last_result = None
-    while asyncio.get_event_loop().time() < deadline:
-        last_result = condition()
-        if last_result:
-            return last_result
-        await asyncio.sleep(interval)
-    raise AssertionError(f"{message} (last result: {last_result})")
-
 
 def make_per_participant_e2ee_options() -> rtc.E2EEOptions:
     options = rtc.E2EEOptions()
@@ -126,7 +86,7 @@ async def publish_dummy_video(source: rtc.VideoSource, stop_event: asyncio.Event
 
 
 @pytest.mark.asyncio
-@skip_if_no_credentials()  # type: ignore[untyped-decorator]
+@skip_if_no_credentials()
 async def test_e2ee_per_participant() -> None:
     """E2E test for per-participant E2EE keys.
 

livekit-rtc/tests/test_e2ee_shared_key.py

@@ -18,64 +18,24 @@
 
 import asyncio
 import os
-import uuid
-from typing import Any, Callable, TypeVar
 
 import pytest
 
-from livekit import api, rtc
+from livekit import rtc
+
+from utils import (
+    assert_eventually,
+    create_token,
+    skip_if_no_credentials,
+    unique_room_name,
+)
 
 
 SHARED_KEY = b"12345678"
 WRONG_KEY = b"wrongkey"
 WIDTH, HEIGHT = 320, 180
 FRAME_RATE = 15
 
-T = TypeVar("T")
-
-
-def skip_if_no_credentials() -> Any:
-    required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
-    missing = [var for var in required_vars if not os.getenv(var)]
-    return pytest.mark.skipif(
-        bool(missing), reason=f"Missing environment variables: {', '.join(missing)}"
-    )
-
-
-def create_token(identity: str, room_name: str) -> str:
-    return (
-        api.AccessToken()
-        .with_identity(identity)
-        .with_name(identity)
-        .with_grants(
-            api.VideoGrants(
-                room_join=True,
-                room=room_name,
-            )
-        )
-        .to_jwt()
-    )
-
-
-def unique_room_name(base: str) -> str:
-    return f"{base}-{uuid.uuid4().hex[:8]}"
-
-
-async def assert_eventually(
-    condition: Callable[[], T],
-    timeout: float = 10.0,
-    interval: float = 0.1,
-    message: str = "Condition not met within timeout",
-) -> T:
-    deadline = asyncio.get_event_loop().time() + timeout
-    last_result = None
-    while asyncio.get_event_loop().time() < deadline:
-        last_result = condition()
-        if last_result:
-            return last_result
-        await asyncio.sleep(interval)
-    raise AssertionError(f"{message} (last result: {last_result})")
-
 
 def make_e2ee_options() -> rtc.E2EEOptions:
     options = rtc.E2EEOptions()
@@ -105,7 +65,7 @@ async def publish_dummy_video(source: rtc.VideoSource, stop_event: asyncio.Event
 
 
 @pytest.mark.asyncio
-@skip_if_no_credentials()  # type: ignore[untyped-decorator]
+@skip_if_no_credentials()
 async def test_e2ee_shared_key() -> None:
     """E2E test for shared-key E2EE.
 

livekit-rtc/tests/test_video.py

@@ -14,17 +14,19 @@
 
 """End-to-end video publish/subscribe tests."""
 
+from __future__ import annotations
+
 import asyncio
 import os
 import struct
-import uuid
 import zlib
 from pathlib import Path
 
 import numpy as np
-import pytest
 
-from livekit import api, rtc
+from livekit import rtc
+
+from utils import create_token, skip_if_no_credentials, unique_room_name
 
 
 VIDEO_WIDTH = 640
@@ -41,33 +43,6 @@
 ]
 
 
-def skip_if_no_credentials():
-    required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
-    missing = [var for var in required_vars if not os.getenv(var)]
-    return pytest.mark.skipif(
-        bool(missing), reason=f"Missing environment variables: {', '.join(missing)}"
-    )
-
-
-def create_token(identity: str, room_name: str) -> str:
-    return (
-        api.AccessToken()
-        .with_identity(identity)
-        .with_name(identity)
-        .with_grants(
-            api.VideoGrants(
-                room_join=True,
-                room=room_name,
-            )
-        )
-        .to_jwt()
-    )
-
-
-def unique_room_name(base: str) -> str:
-    return f"{base}-{uuid.uuid4().hex[:8]}"
-
-
 def _solid_color_rgba_frame(width: int, height: int, rgb: tuple[int, int, int]) -> rtc.VideoFrame:
     """Build a solid-color 640x480 RGBA `VideoFrame` for the given RGB triple."""
     pixels = np.empty((height, width, 4), dtype=np.uint8)
@@ -129,7 +104,7 @@ def _chunk(tag: bytes, data: bytes) -> bytes:
 class TestVideoStreamPublishSubscribe:
     """End-to-end: publish a 640x480 color-cycle video and verify colors on the subscriber."""
 
-    async def test_video_stream_publish_subscribe(self):
+    async def test_video_stream_publish_subscribe(self) -> None:
         """Publish red/green/blue/white/black (1s each, 15fps) and verify color sequence."""
         url = os.environ["LIVEKIT_URL"]
         room_name = unique_room_name("test-video-colors")
@@ -148,7 +123,7 @@ def on_track_subscribed(
             track: rtc.Track,
             publication: rtc.RemoteTrackPublication,
             participant: rtc.RemoteParticipant,
-        ):
+        ) -> None:
             nonlocal subscribed_track
             if track.kind == rtc.TrackKind.KIND_VIDEO:
                 subscribed_track = track